Repository: pydantic/pydantic
Pull request: Allow applying serializers and constraints to named/ref schemas
Fixes https://github.com/pydantic/pydantic/issues/5998

This makes some fundamental changes to core schema and json schema building.
In particular:
- Any schemas with a `"ref"` or `"$ref"` we immediately stick into `"definitions"` or `"$defs"` respectively. I added a `resolve_ref_schema` method to `GetCoreSchemaHandler` that is analogous to the one we already had in `GetJsonSchemaHandler`. This means that implementing `__get_pydantic_core_schema__` can be more complex because you're forced to deal with refs explicitly, but it resolves several inconsistencies and bugs.
- I was able to make automatic unpacking of `"ref"` schemas work for the source type itself, which means that if you implement `__get_pydantic_core_schema__` on `Model(BaseModel)` and call `handler(source_type)` you don't get back a `"ref"` schema, we unpack it for you because a source type should edit its own schema in-place.
- We no longer "automatically unpack" `"$ref"` JSON schemas. To avoid making JSON schema generation more difficult we were previously automatically unpacking `"$ref"` for every handler call, but this was complex and led to inconsistencies the same as mentioned above. This leads to regression for the case of "source types" because as per above those should have the `"ref"` unpacked for them. Fixing this will require some more work that I would like to defer from this PR. In particular, when we generate the JSON schema now we just have a list of functions to call, we have no way of knowing if they came from metadata in `Annotated` or `__get_pydantic_json_schema__` on the source type itself.


Selected Reviewer: @dmontagu

Problem statement:
Field serializers clobber serialization for the field type's schema
```python
from pydantic import BaseModel, field_serializer


class User(BaseModel):
    y: str


class Model(BaseModel):
    user1: User
    user2: User

    @field_serializer('user1')
    def serialize_user1(self, v: User):
        return f'User1({v.y})'

    @field_serializer('user2')
    def serialize_user2(self, v: User):
        return f'User2({v.y})'

m = Model(user1=User(y='1'), user2=User(y='2'))
# fails, actual: {'user1': 'User2(1)', 'user2': 'User2(2)'}
assert m.model_dump() == {'user1': 'User1(1)', 'user2': 'User2(2)'} 
```

Hints from issue discussion:

Commit message: Address feedback by adding docstrings and moving class to end of file

Patch to review:
diff --git a/docs/usage/json_schema.md b/docs/usage/json_schema.md
index 0014aeecb68..7536e0622f0 100644
--- a/docs/usage/json_schema.md
+++ b/docs/usage/json_schema.md
@@ -774,6 +774,7 @@ class Person(BaseModel):
         cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler
     ) -> JsonSchemaValue:
         json_schema = handler(core_schema)
+        json_schema = handler.resolve_ref_schema(json_schema)
         json_schema['examples'] = [
             {
                 'name': 'John Doe',
diff --git a/pydantic/_internal/_annotated_handlers.py b/pydantic/_internal/_annotated_handlers.py
index e1ef1542c20..0bfa955dfb4 100644
--- a/pydantic/_internal/_annotated_handlers.py
+++ b/pydantic/_internal/_annotated_handlers.py
@@ -88,3 +88,19 @@ def generate_schema(self, __source_type: Any) -> core_schema.CoreSchema:
             CoreSchema: the `pydantic-core` CoreSchema generated.
         """
         raise NotImplementedError
+
+    def resolve_ref_schema(self, __maybe_ref_schema: core_schema.CoreSchema) -> core_schema.CoreSchema:
+        """Get the real schema for a `definition-ref` schema.
+        If the schema given is not a `definition-ref` schema, it will be returned as is.
+        This means you don't have to check before calling this function.
+
+        Args:
+            __maybe_ref_schema: A `CoreSchema`, `ref`-based or not.
+
+        Raises:
+            LookupError: If the `ref` is not found.
+
+        Returns:
+            A concrete `CoreSchema`.
+        """
+        raise NotImplementedError
diff --git a/pydantic/_internal/_core_utils.py b/pydantic/_internal/_core_utils.py
index fb03cadc1ed..96c64d125a3 100644
--- a/pydantic/_internal/_core_utils.py
+++ b/pydantic/_internal/_core_utils.py
@@ -100,40 +100,6 @@ def get_type_ref(type_: type[Any], args_override: tuple[type[Any], ...] | None =
     return type_ref
 
 
-def consolidate_refs(schema: core_schema.CoreSchema) -> core_schema.CoreSchema:
-    """This function walks a schema recursively, replacing all but the first occurrence of each ref with
-    a definition-ref schema referencing that ref.
-
-    This makes the fundamental assumption that any time two schemas have the same ref, occurrences
-    after the first can safely be replaced.
-
-    In most cases, schemas with the same ref should not actually be produced. However, when building recursive
-    models with multiple references to themselves at some level in the field hierarchy, it is difficult to avoid
-    getting multiple (identical) copies of the same schema with the same ref. This function removes the copied refs,
-    but is safe because the "duplicate" refs refer to the same schema.
-
-    There is one case where we purposely emit multiple (different) schemas with the same ref: when building
-    recursive generic models. In this case, as an implementation detail, recursive generic models will emit
-    a _non_-identical schema deeper in the tree with a re-used ref, with the intent that _that_ schema will
-    be replaced with a recursive reference once the specific generic parametrization to use can be determined.
-    """
-    refs: set[str] = set()
-
-    top_ref = schema.get('ref', None)  # type: ignore[assignment]
-
-    def _replace_refs(s: core_schema.CoreSchema, recurse: Recurse) -> core_schema.CoreSchema:
-        ref: str | None = s.get('ref')  # type: ignore[assignment]
-        if ref:
-            if ref is top_ref:
-                return recurse(s, _replace_refs)
-            if ref in refs:
-                return {'type': 'definition-ref', 'schema_ref': ref}
-            refs.add(ref)
-        return recurse(s, _replace_refs)
-
-    return walk_core_schema(schema, _replace_refs)
-
-
 def collect_definitions(schema: core_schema.CoreSchema) -> dict[str, core_schema.CoreSchema]:
     # Only collect valid definitions. This is equivalent to collecting all definitions for "valid" schemas,
     # but allows us to reuse this logic while removing "invalid" definitions
@@ -453,7 +419,8 @@ def walk_core_schema(schema: core_schema.CoreSchema, f: Walk) -> core_schema.Cor
 
 
 def _simplify_schema_references(schema: core_schema.CoreSchema, inline: bool) -> core_schema.CoreSchema:  # noqa: C901
-    all_defs: dict[str, core_schema.CoreSchema] = {}
+    valid_defs: dict[str, core_schema.CoreSchema] = {}
+    invalid_defs: dict[str, core_schema.CoreSchema] = {}
 
     def make_result(schema: core_schema.CoreSchema, defs: Iterable[core_schema.CoreSchema]) -> core_schema.CoreSchema:
         definitions = list(defs)
@@ -469,25 +436,39 @@ def collect_refs(s: core_schema.CoreSchema, recurse: Recurse) -> core_schema.Cor
             for definition in s['definitions']:
                 ref = get_ref(definition)
                 assert ref is not None
-                all_defs[ref] = recurse(definition, collect_refs).copy()
+                def_schema = recurse(definition, collect_refs).copy()
+                if 'invalid' in def_schema.get('metadata', {}):
+                    invalid_defs[ref] = def_schema
+                else:
+                    valid_defs[ref] = def_schema
             return recurse(s['schema'], collect_refs)
         ref = get_ref(s)
         if ref is not None:
-            all_defs[ref] = s
+            if 'invalid' in s.get('metadata', {}):
+                invalid_defs[ref] = s
+            else:
+                valid_defs[ref] = s
         return recurse(s, collect_refs)
 
     schema = walk_core_schema(schema, collect_refs)
 
+    all_defs = {**invalid_defs, **valid_defs}
+
     def flatten_refs(s: core_schema.CoreSchema, recurse: Recurse) -> core_schema.CoreSchema:
         if is_definitions_schema(s):
+            new: dict[str, Any] = dict(s)
             # iterate ourselves, we don't want to flatten the actual defs!
-            s['schema'] = recurse(s['schema'], flatten_refs)
-            for definition in s['definitions']:
+            definitions: list[CoreSchema] = new.pop('definitions')
+            schema = cast(CoreSchema, new.pop('schema'))
+            # remaining keys are optional like 'serialization'
+            schema = cast(CoreSchema, {**schema, **new})
+            s['schema'] = recurse(schema, flatten_refs)
+            for definition in definitions:
                 recurse(definition, flatten_refs)  # don't re-assign here!
-            return s
+            return schema
         s = recurse(s, flatten_refs)
         ref = get_ref(s)
-        if ref and ref in all_defs and ref:
+        if ref and ref in all_defs:
             all_defs[ref] = s
             return core_schema.definition_reference_schema(schema_ref=ref)
         return s
@@ -528,6 +509,10 @@ def inline_refs(s: core_schema.CoreSchema, recurse: Recurse) -> core_schema.Core
                 new = all_defs.pop(ref)
                 ref_counts[ref] -= 1  # because we just replaced it!
                 new.pop('ref')  # type: ignore
+                # put all other keys that were on the def-ref schema into the inlined version
+                # in particular this is needed for `serialization`
+                if 'serialization' in s:
+                    new['serialization'] = s['serialization']
                 s = recurse(new, inline_refs)
                 return s
             else:
diff --git a/pydantic/_internal/_dataclasses.py b/pydantic/_internal/_dataclasses.py
index 15644d4106c..74cbfb85026 100644
--- a/pydantic/_internal/_dataclasses.py
+++ b/pydantic/_internal/_dataclasses.py
@@ -98,7 +98,8 @@ def __init__(__dataclass_self__: PydanticDataclass, *args: Any, **kwargs: Any) -
                 cls,
                 CallbackGetCoreSchemaHandler(
                     partial(gen_schema.generate_schema, from_dunder_get_core_schema=False),
-                    gen_schema.generate_schema,
+                    gen_schema,
+                    ref_mode='unpack',
                 ),
             )
         else:
@@ -128,6 +129,8 @@ def attempt_rebuild() -> SchemaValidator | None:
 
     core_config = config_wrapper.core_config(cls)
 
+    schema = gen_schema.collect_definitions(schema)
+
     # We are about to set all the remaining required properties expected for this cast;
     # __pydantic_decorators__ and __pydantic_fields__ should already be set
     cls = typing.cast('type[PydanticDataclass]', cls)
diff --git a/pydantic/_internal/_discriminated_union.py b/pydantic/_internal/_discriminated_union.py
index 17c71e8e697..be014ddb0c6 100644
--- a/pydantic/_internal/_discriminated_union.py
+++ b/pydantic/_internal/_discriminated_union.py
@@ -88,6 +88,7 @@ def apply(self, schema: core_schema.CoreSchema) -> core_schema.CoreSchema:
         to this class.
         """
         old_definitions = collect_definitions(schema)
+        self.definitions = {**old_definitions, **self.definitions}
         assert not self._used
         schema = self._apply_to_root(schema)
         if self._should_be_nullable and not self._is_nullable:
diff --git a/pydantic/_internal/_generate_schema.py b/pydantic/_internal/_generate_schema.py
index 162dc21cb6e..0b943a33725 100644
--- a/pydantic/_internal/_generate_schema.py
+++ b/pydantic/_internal/_generate_schema.py
@@ -7,12 +7,14 @@
 import re
 import sys
 import typing
+from contextlib import contextmanager
+from copy import copy
 from functools import partial
 from inspect import Parameter, _ParameterKind, signature
 from itertools import chain
 from operator import attrgetter
 from types import FunctionType, LambdaType, MethodType
-from typing import TYPE_CHECKING, Any, Callable, ForwardRef, Iterable, Mapping, TypeVar, Union, cast
+from typing import TYPE_CHECKING, Any, Callable, ForwardRef, Iterable, Iterator, Mapping, TypeVar, Union, cast
 
 from pydantic_core import CoreSchema, core_schema
 from typing_extensions import Annotated, Final, Literal, TypeAliasType, TypedDict, get_args, get_origin, is_typeddict
@@ -30,7 +32,6 @@
 )
 from ._core_utils import (
     CoreSchemaOrField,
-    consolidate_refs,
     define_expected_missing_refs,
     get_type_ref,
     is_list_like_schema_with_items_schema,
@@ -59,8 +60,6 @@
 from ._generics import get_standard_typevars_map, recursively_defined_type_refs, replace_types
 from ._schema_generation_shared import (
     CallbackGetCoreSchemaHandler,
-    UnpackedRefJsonSchemaHandler,
-    wrap_json_schema_fn_for_model_or_custom_type_with_ref_unpacking,
 )
 from ._typing_extra import is_finalvar
 from ._utils import lenient_issubclass
@@ -154,10 +153,9 @@ def modify_model_json_schema(
     schema_or_field: CoreSchemaOrField, handler: GetJsonSchemaHandler, *, cls: Any
 ) -> JsonSchemaValue:
     """Add title and description for model-like classes' JSON schema."""
-    wrapped_handler = UnpackedRefJsonSchemaHandler(handler)
 
     json_schema = handler(schema_or_field)
-    original_schema = wrapped_handler.resolve_ref_schema(json_schema)
+    original_schema = handler.resolve_ref_schema(json_schema)
     if 'title' not in original_schema:
         original_schema['title'] = cls.__name__
     docstring = cls.__doc__
@@ -167,7 +165,7 @@ def modify_model_json_schema(
 
 
 class GenerateSchema:
-    __slots__ = '_config_wrapper_stack', 'types_namespace', 'typevars_map', 'recursion_cache', 'definitions'
+    __slots__ = '_config_wrapper_stack', 'types_namespace', 'typevars_map', 'recursion_cache', 'definitions', 'defs'
 
     def __init__(
         self,
@@ -180,8 +178,7 @@ def __init__(
         self.types_namespace = types_namespace
         self.typevars_map = typevars_map
 
-        self.recursion_cache: dict[str, core_schema.DefinitionReferenceSchema] = {}
-        self.definitions: dict[str, core_schema.CoreSchema] = {}
+        self.defs = _Definitions()
 
     @property
     def config_wrapper(self) -> ConfigWrapper:
@@ -191,6 +188,15 @@ def config_wrapper(self) -> ConfigWrapper:
     def arbitrary_types(self) -> bool:
         return self.config_wrapper.arbitrary_types_allowed
 
+    def collect_definitions(self, schema: CoreSchema) -> CoreSchema:
+        ref = cast('str | None', schema.get('ref', None))
+        if ref:
+            self.defs.definitions[ref] = schema
+        return core_schema.definitions_schema(
+            schema,
+            list(self.defs.definitions.values()),
+        )
+
     def generate_schema(
         self,
         obj: Any,
@@ -224,86 +230,78 @@ def _generate_schema_for_type(
 
         metadata_js_function = _extract_get_pydantic_json_schema(obj, schema)
         if metadata_js_function is not None:
-            metadata = CoreMetadataHandler(schema).metadata
-            metadata.setdefault('pydantic_js_functions', []).append(metadata_js_function)
-
-        schema = remove_unnecessary_invalid_definitions(schema)
-
-        ref = schema.get('ref', None)
-        if ref:
-            # definitions and definition-ref schemas don't have 'ref', causing the type error ignored on the next line
-            self.definitions[ref] = schema
-
-        return schema
-
-    def model_schema(self, cls: type[BaseModel]) -> core_schema.CoreSchema:
-        """Generate schema for a pydantic model.
-
-        Since models generate schemas for themselves this method is public and can be called
-        from within BaseModel's metaclass.
-        """
-        model_ref, schema = self._get_or_cache_recursive_ref(cls)
-        if schema is not None:
-            return schema
-
-        fields = cls.model_fields
-        decorators = cls.__pydantic_decorators__
-        check_decorator_fields_exist(
-            chain(
-                decorators.field_validators.values(),
-                decorators.field_serializers.values(),
-                decorators.validators.values(),
-            ),
-            fields.keys(),
-        )
-        config_wrapper = ConfigWrapper(cls.model_config, check=False)
-        core_config = config_wrapper.core_config(cls)
-        metadata = build_metadata_dict(js_functions=[partial(modify_model_json_schema, cls=cls)])
-
-        model_validators = decorators.model_validators.values()
-
-        if cls.__pydantic_root_model__:
-            root_field = self._common_field_schema('root', fields['root'], decorators)
-            inner_schema = root_field['schema']
-            inner_schema = apply_model_validators(inner_schema, model_validators, 'inner')
-            model_schema = core_schema.model_schema(
-                cls,
-                inner_schema,
-                custom_init=getattr(cls, '__pydantic_custom_init__', None),
-                root_model=True,
-                post_init=getattr(cls, '__pydantic_post_init__', None),
-                config=core_config,
-                ref=model_ref,
-                metadata={**metadata, **root_field['metadata']},
+            metadata_schema = resolve_original_schema(schema, self.defs.definitions)
+            if metadata_schema:
+                metadata = CoreMetadataHandler(metadata_schema).metadata
+                metadata.setdefault('pydantic_js_functions', []).append(metadata_js_function)
+
+        return remove_unnecessary_invalid_definitions(schema)
+
+    def _model_schema(self, cls: type[BaseModel]) -> core_schema.CoreSchema:
+        """Generate schema for a Pydantic model."""
+        with self.defs.get_schema_or_ref(cls) as (model_ref, maybe_schema):
+            if maybe_schema is not None:
+                return maybe_schema
+
+            fields = cls.model_fields
+            decorators = cls.__pydantic_decorators__
+            check_decorator_fields_exist(
+                chain(
+                    decorators.field_validators.values(),
+                    decorators.field_serializers.values(),
+                    decorators.validators.values(),
+                ),
+                fields.keys(),
             )
-        else:
-            self._config_wrapper_stack.append(config_wrapper)
-            try:
-                fields_schema: core_schema.CoreSchema = core_schema.model_fields_schema(
-                    {k: self._generate_md_field_schema(k, v, decorators) for k, v in fields.items()},
-                    computed_fields=[self._computed_field_schema(d) for d in decorators.computed_fields.values()],
+            config_wrapper = ConfigWrapper(cls.model_config, check=False)
+            core_config = config_wrapper.core_config(cls)
+            metadata = build_metadata_dict(js_functions=[partial(modify_model_json_schema, cls=cls)])
+
+            model_validators = decorators.model_validators.values()
+
+            if cls.__pydantic_root_model__:
+                root_field = self._common_field_schema('root', fields['root'], decorators)
+                inner_schema = root_field['schema']
+                inner_schema = apply_model_validators(inner_schema, model_validators, 'inner')
+                model_schema = core_schema.model_schema(
+                    cls,
+                    inner_schema,
+                    custom_init=getattr(cls, '__pydantic_custom_init__', None),
+                    root_model=True,
+                    post_init=getattr(cls, '__pydantic_post_init__', None),
+                    config=core_config,
+                    ref=model_ref,
+                    metadata={**metadata, **root_field['metadata']},
+                )
+            else:
+                self._config_wrapper_stack.append(config_wrapper)
+                try:
+                    fields_schema: core_schema.CoreSchema = core_schema.model_fields_schema(
+                        {k: self._generate_md_field_schema(k, v, decorators) for k, v in fields.items()},
+                        computed_fields=[self._computed_field_schema(d) for d in decorators.computed_fields.values()],
+                    )
+                finally:
+                    self._config_wrapper_stack.pop()
+
+                inner_schema = apply_validators(fields_schema, decorators.root_validators.values())
+                inner_schema = define_expected_missing_refs(inner_schema, recursively_defined_type_refs())
+                inner_schema = apply_model_validators(inner_schema, model_validators, 'inner')
+
+                model_schema = core_schema.model_schema(
+                    cls,
+                    inner_schema,
+                    custom_init=getattr(cls, '__pydantic_custom_init__', None),
+                    root_model=False,
+                    post_init=getattr(cls, '__pydantic_post_init__', None),
+                    config=core_config,
+                    ref=model_ref,
+                    metadata=metadata,
                 )
-            finally:
-                self._config_wrapper_stack.pop()
-
-            inner_schema = apply_validators(fields_schema, decorators.root_validators.values())
-            inner_schema = define_expected_missing_refs(inner_schema, recursively_defined_type_refs())
-            inner_schema = apply_model_validators(inner_schema, model_validators, 'inner')
-
-            model_schema = core_schema.model_schema(
-                cls,
-                inner_schema,
-                custom_init=getattr(cls, '__pydantic_custom_init__', None),
-                root_model=False,
-                post_init=getattr(cls, '__pydantic_post_init__', None),
-                config=core_config,
-                ref=model_ref,
-                metadata=metadata,
-            )
 
-        model_schema = consolidate_refs(model_schema)
-        schema = self._apply_model_serializers(model_schema, decorators.model_serializers.values())
-        return apply_model_validators(schema, model_validators, 'outer')
+            schema = self._apply_model_serializers(model_schema, decorators.model_serializers.values())
+            schema = apply_model_validators(schema, model_validators, 'outer')
+            self.defs.definitions[model_ref] = schema
+            return core_schema.definition_reference_schema(model_ref)
 
     def _generate_schema_from_prepare_annotations(self, obj: Any) -> core_schema.CoreSchema | None:
         """Try to generate schema from either the `__get_pydantic_core_schema__` function or
@@ -328,6 +326,15 @@ def _generate_schema_from_property(self, obj: Any, source: Any) -> core_schema.C
         Note: `__get_pydantic_core_schema__` takes priority so it can
         decide whether to use a `__pydantic_core_schema__` attribute, or generate a fresh schema.
         """
+        # avoid calling `__get_pydantic_core_schema__` if we've already visited this object
+        with self.defs.get_schema_or_ref(obj) as (_, maybe_schema):
+            if maybe_schema is not None:
+                return maybe_schema
+        if obj is source:
+            ref_mode = 'unpack'
+        else:
+            ref_mode = 'to-def'
+
         get_schema = getattr(obj, '__get_pydantic_core_schema__', None)
         if get_schema is None:
             return None
@@ -336,7 +343,11 @@ def _generate_schema_from_property(self, obj: Any, source: Any) -> core_schema.C
             # (source) -> CoreSchema
             return get_schema(source)
 
-        return get_schema(source, CallbackGetCoreSchemaHandler(self._generate_schema, self.generate_schema))
+        schema = get_schema(source, CallbackGetCoreSchemaHandler(self._generate_schema, self, ref_mode=ref_mode))
+        if 'ref' in schema:
+            self.defs.definitions[schema['ref']] = schema
+            return core_schema.definition_reference_schema(schema['ref'])
+        return schema
 
     def _generate_schema(self, obj: Any) -> core_schema.CoreSchema:  # noqa: C901
         """Recursively generate a pydantic-core schema for any supported python type."""
@@ -372,7 +383,7 @@ def _generate_schema(self, obj: Any) -> core_schema.CoreSchema:  # noqa: C901
         from ..main import BaseModel
 
         if lenient_issubclass(obj, BaseModel):
-            return self.model_schema(obj)
+            return self._model_schema(obj)
 
         if isinstance(obj, PydanticRecursiveRef):
             return core_schema.definition_reference_schema(schema_ref=obj.type_ref)
@@ -500,8 +511,6 @@ def _generate_md_field_schema(
         name: str,
         field_info: FieldInfo,
         decorators: DecoratorInfos,
-        *,
-        required: bool = True,
     ) -> core_schema.ModelField:
         """Prepare a ModelField to represent a model field."""
         common_field = self._common_field_schema(name, field_info, decorators)
@@ -539,7 +548,9 @@ def _common_field_schema(self, name: str, field_info: FieldInfo, decorators: Dec
 
         def apply_discriminator(schema: CoreSchema) -> CoreSchema:
             if field_info.discriminator is not None:
-                schema = _discriminated_union.apply_discriminator(schema, field_info.discriminator, self.definitions)
+                schema = _discriminated_union.apply_discriminator(
+                    schema, field_info.discriminator, self.defs.definitions
+                )
             return schema
 
         source_type, annotations = field_info.annotation, field_info.metadata
@@ -637,31 +648,24 @@ def _type_alias_type_schema(
         obj: Any,  # TypeAliasType
     ) -> CoreSchema:
         origin = get_origin(obj)
-        if origin is not None and _typing_extra.origin_is_type_alias_type(origin):  # type: ignore
-            origin = cast(Any, origin)
-            ref, schema = self._get_or_cache_recursive_ref(origin)
-            if schema is not None:
-                return schema
+        origin = origin or obj
+        with self.defs.get_schema_or_ref(origin) as (ref, maybe_schema):
+            if maybe_schema is not None:
+                return maybe_schema
+
             namespace = (self.types_namespace or {}).copy()
             new_namespace = {**_typing_extra.get_cls_types_namespace(origin), **namespace}
             annotation = origin.__value__
-        else:
-            ref, schema = self._get_or_cache_recursive_ref(obj)
-            if schema is not None:
-                return schema
-            namespace = (self.types_namespace or {}).copy()
-            new_namespace = {**_typing_extra.get_cls_types_namespace(obj), **namespace}
-            annotation = obj.__value__
-        self.types_namespace = new_namespace
-        typevars_map = get_standard_typevars_map(obj)
-        annotation = replace_types(annotation, typevars_map)
-        schema = self.generate_schema(annotation)
-        assert schema['type'] != 'definitions'
-        schema['ref'] = ref  # type: ignore
-        self.types_namespace = namespace or None
-        self.recursion_cache[obj] = schema  # type: ignore
-        self.definitions[ref] = schema
-        return schema
+
+            self.types_namespace = new_namespace
+            typevars_map = get_standard_typevars_map(obj)
+            annotation = replace_types(annotation, typevars_map)
+            schema = self.generate_schema(annotation)
+            assert schema['type'] != 'definitions'
+            schema['ref'] = ref  # type: ignore
+            self.types_namespace = namespace or None
+            self.defs.definitions[ref] = schema
+            return core_schema.definition_reference_schema(ref)
 
     def _literal_schema(self, literal_type: Any) -> core_schema.LiteralSchema:
         """Generate schema for a Literal."""
@@ -682,59 +686,63 @@ def _typed_dict_schema(self, typed_dict_cls: Any, origin: Any) -> core_schema.Co
         Hence to avoid creating validators that do not do what users expect we only
         support typing.TypedDict on Python >= 3.11 or typing_extension.TypedDict on all versions
         """
-        typed_dict_ref, schema = self._get_or_cache_recursive_ref(typed_dict_cls)
-        if schema is not None:
-            return schema
+        with self.defs.get_schema_or_ref(typed_dict_cls) as (typed_dict_ref, maybe_schema):
+            if maybe_schema is not None:
+                return maybe_schema
 
-        typevars_map = get_standard_typevars_map(typed_dict_cls)
-        if origin is not None:
-            typed_dict_cls = origin
+            typevars_map = get_standard_typevars_map(typed_dict_cls)
+            if origin is not None:
+                typed_dict_cls = origin
 
-        if not _SUPPORTS_TYPEDDICT and type(typed_dict_cls).__module__ == 'typing':
-            raise PydanticUserError(
-                'Please use `typing_extensions.TypedDict` instead of `typing.TypedDict` on Python < 3.12.',
-                code='typed-dict-version',
-            )
+            if not _SUPPORTS_TYPEDDICT and type(typed_dict_cls).__module__ == 'typing':
+                raise PydanticUserError(
+                    'Please use `typing_extensions.TypedDict` instead of `typing.TypedDict` on Python < 3.12.',
+                    code='typed-dict-version',
+                )
 
-        config: ConfigDict | None = getattr(typed_dict_cls, '__pydantic_config__', None)
-        config_wrapper = ConfigWrapper(config)
-        core_config = config_wrapper.core_config(None)
+            config: ConfigDict | None = getattr(typed_dict_cls, '__pydantic_config__', None)
+            config_wrapper = ConfigWrapper(config)
+            core_config = config_wrapper.core_config(None)
 
-        required_keys: frozenset[str] = typed_dict_cls.__required_keys__
+            required_keys: frozenset[str] = typed_dict_cls.__required_keys__
 
-        fields: dict[str, core_schema.TypedDictField] = {}
+            fields: dict[str, core_schema.TypedDictField] = {}
 
-        decorators = DecoratorInfos.build(typed_dict_cls)
+            decorators = DecoratorInfos.build(typed_dict_cls)
 
-        for field_name, annotation in get_type_hints_infer_globalns(
-            typed_dict_cls, localns=self.types_namespace, include_extras=True
-        ).items():
-            annotation = replace_types(annotation, typevars_map)
-            required = field_name in required_keys
+            for field_name, annotation in get_type_hints_infer_globalns(
+                typed_dict_cls, localns=self.types_namespace, include_extras=True
+            ).items():
+                annotation = replace_types(annotation, typevars_map)
+                required = field_name in required_keys
 
-            if get_origin(annotation) == _typing_extra.Required:
-                required = True
-                annotation = get_args(annotation)[0]
-            elif get_origin(annotation) == _typing_extra.NotRequired:
-                required = False
-                annotation = get_args(annotation)[0]
+                if get_origin(annotation) == _typing_extra.Required:
+                    required = True
+                    annotation = get_args(annotation)[0]
+                elif get_origin(annotation) == _typing_extra.NotRequired:
+                    required = False
+                    annotation = get_args(annotation)[0]
 
-            field_info = FieldInfo.from_annotation(annotation)
-            fields[field_name] = self._generate_td_field_schema(field_name, field_info, decorators, required=required)
+                field_info = FieldInfo.from_annotation(annotation)
+                fields[field_name] = self._generate_td_field_schema(
+                    field_name, field_info, decorators, required=required
+                )
 
-        metadata = build_metadata_dict(js_functions=[partial(modify_model_json_schema, cls=typed_dict_cls)])
+            metadata = build_metadata_dict(js_functions=[partial(modify_model_json_schema, cls=typed_dict_cls)])
 
-        td_schema = core_schema.typed_dict_schema(
-            fields,
-            computed_fields=[self._computed_field_schema(d) for d in decorators.computed_fields.values()],
-            extra_behavior='forbid',
-            ref=typed_dict_ref,
-            metadata=metadata,
-            config=core_config,
-        )
+            td_schema = core_schema.typed_dict_schema(
+                fields,
+                computed_fields=[self._computed_field_schema(d) for d in decorators.computed_fields.values()],
+                extra_behavior='forbid',
+                ref=typed_dict_ref,
+                metadata=metadata,
+                config=core_config,
+            )
 
-        schema = self._apply_model_serializers(td_schema, decorators.model_serializers.values())
-        return apply_model_validators(schema, decorators.model_validators.values(), 'all')
+            schema = self._apply_model_serializers(td_schema, decorators.model_serializers.values())
+            schema = apply_model_validators(schema, decorators.model_validators.values(), 'all')
+            self.defs.definitions[typed_dict_ref] = schema
+            return core_schema.definition_reference_schema(typed_dict_ref)
 
     def _namedtuple_schema(self, namedtuple_cls: Any) -> core_schema.CallSchema:
         """Generate schema for a NamedTuple."""
@@ -902,72 +910,74 @@ def _dataclass_schema(
         self, dataclass: type[StandardDataclass], origin: type[StandardDataclass] | None
     ) -> core_schema.CoreSchema:
         """Generate schema for a dataclass."""
-        dataclass_ref, schema = self._get_or_cache_recursive_ref(dataclass)
-        if schema is not None:
-            return schema
+        with self.defs.get_schema_or_ref(dataclass) as (dataclass_ref, maybe_schema):
+            if maybe_schema is not None:
+                return maybe_schema
 
-        typevars_map = get_standard_typevars_map(dataclass)
-        if origin is not None:
-            dataclass = origin
+            typevars_map = get_standard_typevars_map(dataclass)
+            if origin is not None:
+                dataclass = origin
 
-        from ._dataclasses import is_pydantic_dataclass
+            from ._dataclasses import is_pydantic_dataclass
 
-        if is_pydantic_dataclass(dataclass):
-            fields = dataclass.__pydantic_fields__
-            if typevars_map:
-                for field in fields.values():
-                    field.apply_typevars_map(typevars_map, self.types_namespace)
-        else:
-            fields = collect_dataclass_fields(
-                dataclass,
-                self.types_namespace,
-                typevars_map=typevars_map,
+            if is_pydantic_dataclass(dataclass):
+                fields = dataclass.__pydantic_fields__
+                if typevars_map:
+                    for field in fields.values():
+                        field.apply_typevars_map(typevars_map, self.types_namespace)
+            else:
+                fields = collect_dataclass_fields(
+                    dataclass,
+                    self.types_namespace,
+                    typevars_map=typevars_map,
+                )
+            decorators = dataclass.__dict__.get('__pydantic_decorators__') or DecoratorInfos.build(dataclass)
+            # Move kw_only=False args to the start of the list, as this is how vanilla dataclasses work.
+            # Note that when kw_only is missing or None, it is treated as equivalent to kw_only=True
+            args = sorted(
+                (self._generate_dc_field_schema(k, v, decorators) for k, v in fields.items()),
+                key=lambda a: a.get('kw_only') is not False,
             )
-        decorators = dataclass.__dict__.get('__pydantic_decorators__') or DecoratorInfos.build(dataclass)
-        # Move kw_only=False args to the start of the list, as this is how vanilla dataclasses work.
-        # Note that when kw_only is missing or None, it is treated as equivalent to kw_only=True
-        args = sorted(
-            (self._generate_dc_field_schema(k, v, decorators) for k, v in fields.items()),
-            key=lambda a: a.get('kw_only') is not False,
-        )
-        has_post_init = hasattr(dataclass, '__post_init__')
-        has_slots = hasattr(dataclass, '__slots__')
-
-        config = getattr(dataclass, '__pydantic_config__', None)
-        if config is not None:
-            config_wrapper = ConfigWrapper(config, check=False)
-            self._config_wrapper_stack.append(config_wrapper)
-            core_config = config_wrapper.core_config(dataclass)
-        else:
-            core_config = None
+            has_post_init = hasattr(dataclass, '__post_init__')
+            has_slots = hasattr(dataclass, '__slots__')
 
-        try:
-            args_schema = core_schema.dataclass_args_schema(
-                dataclass.__name__,
-                args,
-                computed_fields=[self._computed_field_schema(d) for d in decorators.computed_fields.values()],
-                collect_init_only=has_post_init,
-            )
-        finally:
+            config = getattr(dataclass, '__pydantic_config__', None)
             if config is not None:
-                self._config_wrapper_stack.pop()
+                config_wrapper = ConfigWrapper(config, check=False)
+                self._config_wrapper_stack.append(config_wrapper)
+                core_config = config_wrapper.core_config(dataclass)
+            else:
+                core_config = None
 
-        inner_schema = apply_validators(args_schema, decorators.root_validators.values())
+            try:
+                args_schema = core_schema.dataclass_args_schema(
+                    dataclass.__name__,
+                    args,
+                    computed_fields=[self._computed_field_schema(d) for d in decorators.computed_fields.values()],
+                    collect_init_only=has_post_init,
+                )
+            finally:
+                if config is not None:
+                    self._config_wrapper_stack.pop()
 
-        model_validators = decorators.model_validators.values()
-        inner_schema = apply_model_validators(inner_schema, model_validators, 'inner')
+            inner_schema = apply_validators(args_schema, decorators.root_validators.values())
 
-        dc_schema = core_schema.dataclass_schema(
-            dataclass,
-            inner_schema,
-            post_init=has_post_init,
-            ref=dataclass_ref,
-            fields=[field.name for field in dataclasses.fields(dataclass)],
-            slots=has_slots,
-            config=core_config,
-        )
-        schema = self._apply_model_serializers(dc_schema, decorators.model_serializers.values())
-        return apply_model_validators(schema, model_validators, 'outer')
+            model_validators = decorators.model_validators.values()
+            inner_schema = apply_model_validators(inner_schema, model_validators, 'inner')
+
+            dc_schema = core_schema.dataclass_schema(
+                dataclass,
+                inner_schema,
+                post_init=has_post_init,
+                ref=dataclass_ref,
+                fields=[field.name for field in dataclasses.fields(dataclass)],
+                slots=has_slots,
+                config=core_config,
+            )
+            schema = self._apply_model_serializers(dc_schema, decorators.model_serializers.values())
+            schema = apply_model_validators(schema, model_validators, 'outer')
+            self.defs.definitions[dataclass_ref] = schema
+            return core_schema.definition_reference_schema(dataclass_ref)
 
     def _callable_schema(self, function: Callable[..., Any]) -> core_schema.CallSchema:
         """Generate schema for a Callable.
@@ -1032,14 +1042,6 @@ def _unsubstituted_typevar_schema(self, typevar: typing.TypeVar) -> core_schema.
         else:
             return core_schema.AnySchema(type='any')
 
-    def _get_or_cache_recursive_ref(self, cls: type[Any]) -> tuple[str, core_schema.DefinitionReferenceSchema | None]:
-        obj_ref = get_type_ref(cls)
-        if obj_ref in self.recursion_cache:
-            return obj_ref, self.recursion_cache[obj_ref]
-        else:
-            self.recursion_cache[obj_ref] = core_schema.definition_reference_schema(obj_ref)
-            return obj_ref, None
-
     def _computed_field_schema(self, d: Decorator[ComputedFieldInfo]) -> core_schema.ComputedField:
         return_type_schema = self.generate_schema(d.info.return_type)
 
@@ -1141,7 +1143,7 @@ def inner_handler(obj: Any) -> CoreSchema:
                     pydantic_js_functions.append(metadata_js_function)
             return transform_inner_schema(schema)
 
-        get_inner_schema = CallbackGetCoreSchemaHandler(inner_handler, self.generate_schema)
+        get_inner_schema = CallbackGetCoreSchemaHandler(inner_handler, self)
 
         while True:
             idx += 1
@@ -1163,20 +1165,62 @@ def inner_handler(obj: Any) -> CoreSchema:
                         annotations,
                     )
             annotation = annotations[idx]
-            get_inner_schema = self._get_wrapped_inner_schema(
-                get_inner_schema, annotation, self.definitions, pydantic_js_functions
-            )
+            get_inner_schema = self._get_wrapped_inner_schema(get_inner_schema, annotation, pydantic_js_functions)
 
         schema = get_inner_schema(source_type)
-        metadata = CoreMetadataHandler(schema).metadata
-        metadata.setdefault('pydantic_js_functions', []).extend(pydantic_js_functions)
+        metadata_schema = resolve_original_schema(schema, self.defs.definitions)
+        if metadata_schema:
+            metadata = CoreMetadataHandler(metadata_schema).metadata
+            metadata.setdefault('pydantic_js_functions', []).extend(pydantic_js_functions)
         return schema
 
+    def apply_single_annotation(self, schema: core_schema.CoreSchema, metadata: Any) -> core_schema.CoreSchema:
+        if isinstance(metadata, FieldInfo):
+            for field_metadata in metadata.metadata:
+                schema = self.apply_single_annotation(schema, field_metadata)
+            if metadata.discriminator is not None:
+                schema = _discriminated_union.apply_discriminator(
+                    schema,
+                    metadata.discriminator,
+                    self.defs.definitions,
+                )
+            return schema
+
+        if schema['type'] == 'nullable':
+            # for nullable schemas, metadata is automatically applied to the inner schema
+            inner = schema.get('schema', core_schema.any_schema())
+            inner = self.apply_single_annotation(inner, metadata)
+            if inner:
+                schema['schema'] = inner
+            return schema
+
+        original_schema = schema
+        ref = schema.get('ref', None)
+        if ref is not None:
+            schema = schema.copy()
+            new_ref = ref + f'_{repr(metadata)}'
+            if new_ref in self.defs.definitions:
+                return self.defs.definitions[new_ref]
+            schema['ref'] = new_ref  # type: ignore
+        elif schema['type'] == 'definition-ref':
+            ref = schema['schema_ref']
+            if ref in self.defs.definitions:
+                schema = self.defs.definitions[ref].copy()
+                new_ref = ref + f'_{repr(metadata)}'
+                if new_ref in self.defs.definitions:
+                    return self.defs.definitions[new_ref]
+                schema['ref'] = new_ref  # type: ignore
+
+        maybe_updated_schema = _known_annotated_metadata.apply_known_metadata(metadata, schema.copy())
+
+        if maybe_updated_schema is not None:
+            return maybe_updated_schema
+        return original_schema
+
     def _get_wrapped_inner_schema(
         self,
         get_inner_schema: GetCoreSchemaHandler,
         annotation: Any,
-        definitions: dict[str, core_schema.CoreSchema],
         pydantic_js_functions: list[GetJsonSchemaFunction],
     ) -> CallbackGetCoreSchemaHandler:
         metadata_get_schema: GetCoreSchemaFunction = getattr(annotation, '__get_pydantic_core_schema__', None) or (
@@ -1185,20 +1229,30 @@ def _get_wrapped_inner_schema(
 
         def new_handler(source: Any) -> core_schema.CoreSchema:
             schema = metadata_get_schema(source, get_inner_schema)
-            schema = apply_single_annotation(schema, annotation, definitions)
+            schema = self.apply_single_annotation(schema, annotation)
 
             metadata_js_function = _extract_get_pydantic_json_schema(annotation, schema)
             if metadata_js_function is not None:
                 pydantic_js_functions.append(metadata_js_function)
             return schema
 
-        return CallbackGetCoreSchemaHandler(new_handler, self.generate_schema)
+        return CallbackGetCoreSchemaHandler(new_handler, self)
 
     def _apply_field_serializers(
         self, schema: core_schema.CoreSchema, serializers: list[Decorator[FieldSerializerDecoratorInfo]]
     ) -> core_schema.CoreSchema:
         """Apply field serializers to a schema."""
         if serializers:
+            schema = copy(schema)
+            if schema['type'] == 'definitions':
+                inner_schema = schema['schema']
+                schema['schema'] = self._apply_field_serializers(inner_schema, serializers)
+                return schema
+            else:
+                ref = typing.cast('str|None', schema.get('ref', None))
+                if ref is not None:
+                    schema = core_schema.definition_reference_schema(ref)
+
             # use the last serializer to make it easy to override a serializer set on a parent model
             serializer = serializers[-1]
             is_field_serializer, info_arg = inspect_field_serializer(serializer.func, serializer.info.mode)
@@ -1355,27 +1409,6 @@ def apply_model_validators(
     return schema
 
 
-def apply_single_annotation(
-    schema: core_schema.CoreSchema, metadata: Any, definitions: dict[str, core_schema.CoreSchema]
-) -> core_schema.CoreSchema:
-    if isinstance(metadata, FieldInfo):
-        for field_metadata in metadata.metadata:
-            schema = apply_single_annotation(schema, field_metadata, definitions)
-        if metadata.discriminator is not None:
-            schema = _discriminated_union.apply_discriminator(schema, metadata.discriminator, definitions)
-        return schema
-
-    if schema['type'] == 'nullable':
-        # for nullable schemas, metadata is automatically applied to the inner schema
-        inner = schema.get('schema', core_schema.any_schema())
-        inner = apply_single_annotation(inner, metadata, definitions)
-        if inner:
-            schema['schema'] = inner
-        return schema
-
-    return _known_annotated_metadata.apply_known_metadata(metadata, schema.copy())
-
-
 def wrap_default(field_info: FieldInfo, schema: core_schema.CoreSchema) -> core_schema.CoreSchema:
     if field_info.default_factory:
         return core_schema.with_default_schema(
@@ -1415,10 +1448,6 @@ def _extract_get_pydantic_json_schema(tp: Any, schema: CoreSchema) -> GetJsonSch
     if js_modify_function is None:
         return None
 
-    # wrap the schema so that we unpack ref schemas and always call metadata_js_function with the full schema
-    if schema['type'] != 'definition-ref':
-        # we would fail to unpack recursive ref schemas!
-        js_modify_function = wrap_json_schema_fn_for_model_or_custom_type_with_ref_unpacking(js_modify_function)
     return js_modify_function
 
 
@@ -1448,3 +1477,50 @@ def _common_field(
         'frozen': frozen,
         'metadata': metadata,
     }
+
+
+class _Definitions:
+    """Keeps track of references and definitions"""
+
+    def __init__(self) -> None:
+        self.seen: set[str] = set()
+        self.definitions: dict[str, core_schema.CoreSchema] = {}
+
+    @contextmanager
+    def get_schema_or_ref(self, tp: Any) -> Iterator[tuple[str, None] | tuple[str, CoreSchema]]:
+        """
+        Get a definition for `tp` if one exists.
+
+        If a definition exists a tuple of (ref_string, CoreSchema) is returned.
+        If no definition exists yet a tuple of (ref_string, None) is returned.
+
+        Note that the returned CoreSchema will always be a DefinitionReferenceSchema,
+        not the actual definition itself.
+
+        This should be called for any type that can be identified by reference.
+        This includes any recursive types.
+        At present the following types can be named / recursive:
+        - BaseModel
+        - Dataclasses
+        - TypedDict
+        - TypeAliasType
+        """
+        ref = get_type_ref(tp)
+        # return the reference if we're either (1) in a cycle or (2) it was already defined
+        if ref in self.seen or ref in self.definitions:
+            yield (ref, core_schema.definition_reference_schema(ref))
+        else:
+            self.seen.add(ref)
+            try:
+                yield (ref, None)
+            finally:
+                self.seen.discard(ref)
+
+
+def resolve_original_schema(schema: CoreSchema, definitions: dict[str, CoreSchema]) -> CoreSchema | None:
+    if schema['type'] == 'definition-ref':
+        return definitions.get(schema['schema_ref'], None)
+    elif schema['type'] == 'definitions':
+        return schema['schema']
+    else:
+        return schema
diff --git a/pydantic/_internal/_known_annotated_metadata.py b/pydantic/_internal/_known_annotated_metadata.py
index f9acfec4049..fbfadd82b72 100644
--- a/pydantic/_internal/_known_annotated_metadata.py
+++ b/pydantic/_internal/_known_annotated_metadata.py
@@ -58,7 +58,7 @@ def expand_grouped_metadata(annotations: Iterable[Any]) -> Iterable[Any]:
             yield annotation
 
 
-def apply_known_metadata(annotation: Any, schema: CoreSchema) -> CoreSchema:  # noqa: C901
+def apply_known_metadata(annotation: Any, schema: CoreSchema) -> CoreSchema | None:  # noqa: C901
     """Apply `annotation` to `schema` if it is an annotation we know about (Gt, Le, etc.).
     Otherwise return `None`.
 
@@ -139,10 +139,12 @@ def val_func(v: Any) -> Any:
             return v
 
         return cs.no_info_after_validator_function(val_func, schema)
-
-    # for all other annotations just update the schema
-    # this includes things like `strict` which apply to pretty much every schema
-    schema.update(schema_update)  # type: ignore
+    elif schema_update:
+        # for all other annotations just update the schema
+        # this includes things like `strict` which apply to pretty much every schema
+        schema.update(schema_update)  # type: ignore
+    else:
+        return None
 
     return schema
 
diff --git a/pydantic/_internal/_model_construction.py b/pydantic/_internal/_model_construction.py
index 783f06b48ef..f6284e4d154 100644
--- a/pydantic/_internal/_model_construction.py
+++ b/pydantic/_internal/_model_construction.py
@@ -362,10 +362,13 @@ def complete_model_class(
         types_namespace,
         typevars_map,
     )
+
     handler = CallbackGetCoreSchemaHandler(
         partial(gen_schema.generate_schema, from_dunder_get_core_schema=False),
-        gen_schema.generate_schema,
+        gen_schema,
+        ref_mode='unpack',
     )
+
     try:
         schema = cls.__get_pydantic_core_schema__(cls, handler)
     except PydanticUndefinedAnnotation as e:
@@ -389,10 +392,11 @@ def attempt_rebuild() -> SchemaValidator | None:
 
     core_config = config_wrapper.core_config(cls)
 
+    schema = gen_schema.collect_definitions(schema)
+    schema = flatten_schema_defs(schema)
+
     # debug(schema)
     cls.__pydantic_core_schema__ = schema
-
-    schema = flatten_schema_defs(schema)
     simplified_core_schema = inline_schema_defs(schema)
     cls.__pydantic_validator__ = SchemaValidator(simplified_core_schema, core_config)
     cls.__pydantic_serializer__ = SchemaSerializer(simplified_core_schema, core_config)
diff --git a/pydantic/_internal/_schema_generation_shared.py b/pydantic/_internal/_schema_generation_shared.py
index 3e56c057422..7716e60c863 100644
--- a/pydantic/_internal/_schema_generation_shared.py
+++ b/pydantic/_internal/_schema_generation_shared.py
@@ -4,69 +4,19 @@
 from typing import TYPE_CHECKING, Any, Callable
 
 from pydantic_core import core_schema
+from typing_extensions import Literal
 
 from ._annotated_handlers import GetCoreSchemaHandler, GetJsonSchemaHandler
 
 if TYPE_CHECKING:
     from ..json_schema import GenerateJsonSchema, JsonSchemaValue
     from ._core_utils import CoreSchemaOrField
+    from ._generate_schema import GenerateSchema
 
     GetJsonSchemaFunction = Callable[[CoreSchemaOrField, GetJsonSchemaHandler], JsonSchemaValue]
     HandlerOverride = Callable[[CoreSchemaOrField], JsonSchemaValue]
 
 
-class UnpackedRefJsonSchemaHandler(GetJsonSchemaHandler):
-    """A GetJsonSchemaHandler implementation that automatically unpacks `$ref`
-    schemas so that the caller doesn't have to worry about that.
-
-    This is used for custom types and models that implement `__get_pydantic_core_schema__`
-    so they always get a `non-$ref` schema.
-
-    Used internally by Pydantic, please do not rely on this implementation.
-    See `GetJsonSchemaHandler` for the handler API.
-    """
-
-    original_schema: JsonSchemaValue | None = None
-
-    def __init__(self, handler: GetJsonSchemaHandler) -> None:
-        self.handler = handler
-        self.mode = handler.mode
-
-    def resolve_ref_schema(self, __maybe_ref_json_schema: JsonSchemaValue) -> JsonSchemaValue:
-        return self.handler.resolve_ref_schema(__maybe_ref_json_schema)
-
-    def __call__(self, core_schema: CoreSchemaOrField) -> JsonSchemaValue:
-        self.original_schema = self.handler(core_schema)
-        return self.resolve_ref_schema(self.original_schema)
-
-    def update_schema(self, schema: JsonSchemaValue) -> JsonSchemaValue:
-        if self.original_schema is None:
-            # handler / our __call__ was never called
-            return schema
-        if '$ref' in self.original_schema:
-            original_referenced_schema = self.resolve_ref_schema(self.original_schema)
-            if schema != original_referenced_schema:
-                # a new schema was returned, update the non-ref schema
-                original_referenced_schema.clear()
-                original_referenced_schema.update(schema)
-            # return self.original_schema, which may be a ref schema
-            return self.original_schema
-        # not a ref schema, return the new schema
-        return schema
-
-
-def wrap_json_schema_fn_for_model_or_custom_type_with_ref_unpacking(
-    fn: GetJsonSchemaFunction,
-) -> GetJsonSchemaFunction:
-    def wrapped(schema_or_field: CoreSchemaOrField, handler: GetJsonSchemaHandler) -> JsonSchemaValue:
-        wrapped_handler = UnpackedRefJsonSchemaHandler(handler)
-        json_schema = fn(schema_or_field, wrapped_handler)
-        json_schema = wrapped_handler.update_schema(json_schema)
-        return json_schema
-
-    return wrapped
-
-
 class GenerateJsonSchemaHandler(GetJsonSchemaHandler):
     """JsonSchemaHandler implementation that doesn't do ref unwrapping by default.
 
@@ -106,13 +56,36 @@ class CallbackGetCoreSchemaHandler(GetCoreSchemaHandler):
     """
 
     def __init__(
-        self, handler: Callable[[Any], core_schema.CoreSchema], generate_schema: Callable[[Any], core_schema.CoreSchema]
+        self,
+        handler: Callable[[Any], core_schema.CoreSchema],
+        generate_schema: GenerateSchema,
+        ref_mode: Literal['to-def', 'unpack'] = 'to-def',
     ) -> None:
         self._handler = handler
         self._generate_schema = generate_schema
+        self._ref_mode = ref_mode
 
     def __call__(self, __source_type: Any) -> core_schema.CoreSchema:
-        return self._handler(__source_type)
+        schema = self._handler(__source_type)
+        ref = schema.get('ref')
+        if self._ref_mode == 'to-def':
+            if ref is not None:
+                self._generate_schema.defs.definitions[ref] = schema
+                return core_schema.definition_reference_schema(ref)
+            return schema
+        else:  # ref_mode = 'unpack
+            return self.resolve_ref_schema(schema)
 
     def generate_schema(self, __source_type: Any) -> core_schema.CoreSchema:
-        return self._generate_schema(__source_type)
+        return self._generate_schema.generate_schema(__source_type)
+
+    def resolve_ref_schema(self, maybe_ref_schema: core_schema.CoreSchema) -> core_schema.CoreSchema:
+        if maybe_ref_schema['type'] != 'definition-ref':
+            return maybe_ref_schema
+        ref = maybe_ref_schema['schema_ref']
+        if ref not in self._generate_schema.defs.definitions:
+            raise LookupError(
+                f'Could not find a ref for {ref}.'
+                ' Maybe you tried to call resolve_ref_schema from within a recursive model?'
+            )
+        return self._generate_schema.defs.definitions[ref]
diff --git a/pydantic/_internal/_std_types_schema.py b/pydantic/_internal/_std_types_schema.py
index 29a690b75d2..3a176e7e46a 100644
--- a/pydantic/_internal/_std_types_schema.py
+++ b/pydantic/_internal/_std_types_schema.py
@@ -99,7 +99,7 @@ def to_enum(__input_value: Any) -> Enum:
     updates = {k: v for k, v in updates.items() if v is not None}
 
     def get_json_schema(_, handler: GetJsonSchemaHandler) -> JsonSchemaValue:
-        json_schema = handler(core_schema.literal_schema([x.value for x in cases]))
+        json_schema = handler(core_schema.literal_schema([x.value for x in cases], ref=enum_ref))
         original_schema = handler.resolve_ref_schema(json_schema)
         update_json_schema(original_schema, updates)
         return json_schema
diff --git a/pydantic/json_schema.py b/pydantic/json_schema.py
index f2bfae02f8c..066ad87f5d8 100644
--- a/pydantic/json_schema.py
+++ b/pydantic/json_schema.py
@@ -181,6 +181,14 @@ def __init__(self, by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLA
 
         self.mode: JsonSchemaMode = 'validation'
 
+        # When we encounter definitions we need to try to build them immediately
+        # so that they are available schemas that reference them
+        # But it's possible that that CoreSchema was never going to be used
+        # (e.g. because the CoreSchema that references short circuits is JSON schema generation without needing
+        #  the reference) so instead of failing altogether if we can't build a definition we
+        # store the error raised and re-throw it if we end up needing that def
+        self.core_defs_invalid_for_json_schema: dict[DefsRef, PydanticInvalidForJsonSchema] = {}
+
     def build_schema_type_to_method(
         self,
     ) -> dict[CoreSchemaOrFieldType, Callable[[CoreSchemaOrField], JsonSchemaValue]]:
@@ -335,6 +343,27 @@ def generate_inner(self, schema: CoreSchemaOrField) -> JsonSchemaValue:
         # Generate the JSON schema, accounting for the json_schema_override and core_schema_override
         metadata_handler = _core_metadata.CoreMetadataHandler(schema)
 
+        def populate_defs(core_schema: CoreSchema, json_schema: JsonSchemaValue) -> JsonSchemaValue:
+            if 'ref' in core_schema:
+                core_ref = CoreRef(core_schema['ref'])  # type: ignore[typeddict-item]
+                defs_ref, ref_json_schema = self.get_cache_defs_ref_schema(core_ref)
+                json_ref = JsonRef(ref_json_schema['$ref'])
+                self.json_to_defs_refs[json_ref] = defs_ref
+                # Replace the schema if it's not a reference to itself
+                # What we want to avoid is having the def be just a ref to itself
+                # which is what would happen if we blindly assigned any
+                if json_schema.get('$ref', None) != json_ref:
+                    self.definitions[defs_ref] = json_schema
+                    self.core_defs_invalid_for_json_schema.pop(defs_ref, None)
+                json_schema = ref_json_schema
+            if '$ref' in json_schema and len(json_schema.keys()) > 1:
+                # technically you can't have any other keys next to a "$ref"
+                # but it's an easy mistake to make and not hard to correct automatically here
+                json_schema = json_schema.copy()
+                ref = json_schema.pop('$ref')
+                json_schema = {'allOf': [{'$ref': ref}], **json_schema}
+            return json_schema
+
         def handler_func(schema_or_field: CoreSchemaOrField) -> JsonSchemaValue:
             """Generate a JSON schema based on the input schema.
 
@@ -353,12 +382,8 @@ def handler_func(schema_or_field: CoreSchemaOrField) -> JsonSchemaValue:
                 json_schema = generate_for_schema_type(schema_or_field)
             else:
                 raise TypeError(f'Unexpected schema type: schema={schema_or_field}')
-            # Populate the definitions
-            if 'ref' in schema:
-                core_ref = CoreRef(schema['ref'])  # type: ignore[typeddict-item]
-                defs_ref, ref_json_schema = self.get_cache_defs_ref_schema(core_ref)
-                self.definitions[defs_ref] = json_schema
-                json_schema = ref_json_schema
+            if _core_utils.is_core_schema(schema_or_field):
+                json_schema = populate_defs(schema_or_field, json_schema)
             return json_schema
 
         current_handler = _schema_generation_shared.GenerateJsonSchemaHandler(self, handler_func)
@@ -370,7 +395,10 @@ def new_handler_func(
                 current_handler: _core_metadata.GetJsonSchemaHandler = current_handler,
                 js_modify_function: _core_metadata.GetJsonSchemaFunction = js_modify_function,
             ) -> JsonSchemaValue:
-                return js_modify_function(schema_or_field, current_handler)
+                json_schema = js_modify_function(schema_or_field, current_handler)
+                if _core_utils.is_core_schema(schema_or_field):
+                    json_schema = populate_defs(schema_or_field, json_schema)
+                return json_schema
 
             current_handler = _schema_generation_shared.GenerateJsonSchemaHandler(self, new_handler_func)
 
@@ -1178,8 +1206,14 @@ def multi_host_url_schema(self, schema: core_schema.MultiHostUrlSchema) -> JsonS
 
     def definitions_schema(self, schema: core_schema.DefinitionsSchema) -> JsonSchemaValue:
         for definition in schema['definitions']:
-            self.generate_inner(definition)
-        return self.generate_inner(schema['schema'])
+            try:
+                self.generate_inner(definition)
+            except PydanticInvalidForJsonSchema as e:
+                core_ref: CoreRef = CoreRef(definition['ref'])  # type: ignore
+                self.core_defs_invalid_for_json_schema[self.get_defs_ref((core_ref, self.mode))] = e
+                continue
+        res = self.generate_inner(schema['schema'])
+        return res
 
     def definition_ref_schema(self, schema: core_schema.DefinitionReferenceSchema) -> JsonSchemaValue:
         core_ref = CoreRef(schema['schema_ref'])
@@ -1388,7 +1422,10 @@ def handle_ref_overrides(self, json_schema: JsonSchemaValue) -> JsonSchemaValue:
         return json_schema
 
     def get_schema_from_definitions(self, json_ref: JsonRef) -> JsonSchemaValue | None:
-        return self.definitions.get(self.json_to_defs_refs[json_ref])
+        def_ref = self.json_to_defs_refs[json_ref]
+        if def_ref in self.core_defs_invalid_for_json_schema:
+            raise self.core_defs_invalid_for_json_schema[def_ref]
+        return self.definitions.get(def_ref, None)
 
     def encode_default(self, dft: Any) -> Any:
         return pydantic_core.to_jsonable_python(dft)
@@ -1465,7 +1502,10 @@ def _add_json_refs(schema: Any) -> None:
                     json_refs[json_ref] += 1
                     if already_visited:
                         return  # prevent recursion on a definition that was already visited
-                    _add_json_refs(self.definitions[self.json_to_defs_refs[json_ref]])
+                    def_ref = self.json_to_defs_refs[json_ref]
+                    if def_ref in self.core_defs_invalid_for_json_schema:
+                        raise self.core_defs_invalid_for_json_schema[def_ref]
+                    _add_json_refs(self.definitions[def_ref])
                 for v in schema.values():
                     _add_json_refs(v)
             elif isinstance(schema, list):
diff --git a/pydantic/type_adapter.py b/pydantic/type_adapter.py
index 46f3463d049..9eeac459d91 100644
--- a/pydantic/type_adapter.py
+++ b/pydantic/type_adapter.py
@@ -75,7 +75,9 @@ def _get_schema(type_: Any, config_wrapper: _config.ConfigWrapper, parent_depth:
     global_ns = sys._getframe(max(parent_depth - 1, 1)).f_globals.copy()
     global_ns.update(local_ns or {})
     gen = _generate_schema.GenerateSchema(config_wrapper, types_namespace=global_ns, typevars_map={})
-    return gen.generate_schema(type_)
+    schema = gen.generate_schema(type_)
+    schema = gen.collect_definitions(schema)
+    return schema
 
 
 def _getattr_no_parents(obj: Any, attribute: str) -> Any:
diff --git a/tests/test_json.py b/tests/test_json.py
index b354ad48735..72bc3de3ae4 100644
--- a/tests/test_json.py
+++ b/tests/test_json.py
@@ -7,7 +7,7 @@
 from enum import Enum
 from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network
 from pathlib import Path
-from typing import Any, Generator, Optional, Pattern
+from typing import Any, Generator, Optional, Pattern, Union
 from uuid import UUID
 
 import pytest
@@ -80,6 +80,7 @@ def test_json_serialization(ser_type, gen_value, json_output):
     config_wrapper = ConfigWrapper({'arbitrary_types_allowed': False})
     gen = GenerateSchema(config_wrapper, None)
     schema = gen.generate_schema(ser_type)
+    schema = gen.collect_definitions(schema)
     serializer = SchemaSerializer(schema)
     assert serializer.to_json(gen_value()) == json_output
 
@@ -359,21 +360,29 @@ class Model(BaseModel):
     assert M(value=1, nested=M(value=2)).model_dump_json(exclude_none=True) == '{"value":1,"nested":{"value":2}}'
 
 
-def test_unresolvable_schema_lookup_error():
+def test_resolve_ref_schema_recursive_model():
     class Model(BaseModel):
-        mini_me: 'Model'
+        mini_me: Union['Model', None]
 
         @classmethod
         def __get_pydantic_json_schema__(
             cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler
         ) -> JsonSchemaValue:
             json_schema = super().__get_pydantic_json_schema__(core_schema, handler)
-            return handler.resolve_ref_schema(json_schema)
-
-    with pytest.raises(LookupError) as e:
-        Model.model_json_schema()
-
-    assert e.value.args[0] == (
-        'Could not find a ref for #/$defs/Model.'
-        ' Maybe you tried to call resolve_ref_schema from within a recursive model?'
-    )
+            json_schema = handler.resolve_ref_schema(json_schema)
+            json_schema['examples'] = {'foo': {'mini_me': None}}
+            return json_schema
+
+    # insert_assert(Model.model_json_schema())
+    assert Model.model_json_schema() == {
+        '$defs': {
+            'Model': {
+                'examples': {'foo': {'mini_me': None}},
+                'properties': {'mini_me': {'anyOf': [{'$ref': '#/$defs/Model'}, {'type': 'null'}]}},
+                'required': ['mini_me'],
+                'title': 'Model',
+                'type': 'object',
+            }
+        },
+        'allOf': [{'$ref': '#/$defs/Model'}],
+    }
diff --git a/tests/test_json_schema.py b/tests/test_json_schema.py
index 368c54a6144..0a5a948fe36 100644
--- a/tests/test_json_schema.py
+++ b/tests/test_json_schema.py
@@ -30,7 +30,7 @@
 from uuid import UUID
 
 import pytest
-from pydantic_core import CoreSchema, core_schema
+from pydantic_core import CoreSchema, core_schema, to_json
 from typing_extensions import Annotated, Literal
 
 from pydantic import (
@@ -248,8 +248,8 @@ class Model(BaseModel):
         bar: BarEnum
         spam: SpamEnum = Field(None)
 
+    # insert_assert(Model.model_json_schema())
     assert Model.model_json_schema() == {
-        'title': 'Model',
         'type': 'object',
         'properties': {
             'foo': {'$ref': '#/$defs/FooEnum'},
@@ -257,10 +257,11 @@ class Model(BaseModel):
             'spam': {'allOf': [{'$ref': '#/$defs/SpamEnum'}], 'default': None},
         },
         'required': ['foo', 'bar'],
+        'title': 'Model',
         '$defs': {
-            'FooEnum': {'title': 'FooEnum', 'enum': ['f', 'b']},
-            'BarEnum': {'title': 'BarEnum', 'type': 'integer', 'enum': [1, 2]},
-            'SpamEnum': {'title': 'SpamEnum', 'type': 'string', 'enum': ['f', 'b']},
+            'BarEnum': {'enum': [1, 2], 'title': 'BarEnum', 'type': 'integer'},
+            'SpamEnum': {'enum': ['f', 'b'], 'title': 'SpamEnum', 'type': 'string'},
+            'FooEnum': {'enum': ['f', 'b'], 'title': 'FooEnum'},
         },
     }
 
@@ -279,15 +280,16 @@ def __get_pydantic_json_schema__(
             cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler
         ) -> JsonSchemaValue:
             field_schema = handler(core_schema)
+            field_schema = handler.resolve_ref_schema(field_schema)
             existing_comment = field_schema.get('$comment', '')
             field_schema['$comment'] = existing_comment + 'comment'  # make sure this function is only called once
-
             field_schema['tsEnumNames'] = [e.name for e in cls]
             return field_schema
 
     class Model(BaseModel):
         spam: Optional[SpamEnum] = Field(None)
 
+    # insert_assert(Model.model_json_schema())
     assert Model.model_json_schema() == {
         '$defs': {
             'SpamEnum': {
@@ -315,31 +317,26 @@ class Model(BaseModel):
         bulbi: FooBarEnum = Field('foo', alias='bulbialias', title='Bulbibulbi!', description='Bulbi is not...')
         cara: FooBarEnum
 
+    # insert_assert(Model.model_json_schema())
     assert Model.model_json_schema() == {
-        '$defs': {
-            'FooBarEnum': {
-                'enum': ['foo', 'bar'],
-                'title': 'FooBarEnum',
-                'type': 'string',
-            }
-        },
+        'type': 'object',
         'properties': {
             'pikalias': {
-                'allOf': [{'$ref': '#/$defs/FooBarEnum'}],
-                'description': 'Pika is definitely the best!',
                 'title': 'Pikapika!',
+                'description': 'Pika is definitely the best!',
+                'allOf': [{'$ref': '#/$defs/FooBarEnum'}],
             },
             'bulbialias': {
                 'allOf': [{'$ref': '#/$defs/FooBarEnum'}],
-                'description': 'Bulbi is not...',
-                'title': 'Bulbibulbi!',
                 'default': 'foo',
+                'title': 'Bulbibulbi!',
+                'description': 'Bulbi is not...',
             },
             'cara': {'$ref': '#/$defs/FooBarEnum'},
         },
         'required': ['pikalias', 'cara'],
         'title': 'Model',
-        'type': 'object',
+        '$defs': {'FooBarEnum': {'enum': ['foo', 'bar'], 'title': 'FooBarEnum', 'type': 'string'}},
     }
 
 
@@ -366,37 +363,34 @@ class Foo(BaseModel):
             description='Description of model',
         )
 
+    # insert_assert(Foo.model_json_schema())
     assert Foo.model_json_schema() == {
-        '$defs': {
-            'Pika': {
-                'properties': {'a': {'title': 'A', 'type': 'string'}},
-                'required': ['a'],
-                'title': 'Pika',
-                'type': 'object',
-            },
-            'Names': {
-                'enum': ['Rick', 'Morty', 'Summer'],
-                'title': 'Names',
-                'type': 'string',
-            },
-        },
+        'type': 'object',
         'properties': {
             'enum': {'$ref': '#/$defs/Names'},
-            'model': {'$ref': '#/$defs/Pika'},
             'titled_enum': {
-                'allOf': [{'$ref': '#/$defs/Names'}],
-                'description': 'Description of enum',
                 'title': 'Title of enum',
+                'description': 'Description of enum',
+                'allOf': [{'$ref': '#/$defs/Names'}],
             },
+            'model': {'$ref': '#/$defs/Pika'},
             'titled_model': {
-                'allOf': [{'$ref': '#/$defs/Pika'}],
-                'description': 'Description of model',
                 'title': 'Title of model',
+                'description': 'Description of model',
+                'allOf': [{'$ref': '#/$defs/Pika'}],
             },
         },
         'required': ['enum', 'titled_enum', 'model', 'titled_model'],
         'title': 'Foo',
-        'type': 'object',
+        '$defs': {
+            'Names': {'enum': ['Rick', 'Morty', 'Summer'], 'title': 'Names', 'type': 'string'},
+            'Pika': {
+                'type': 'object',
+                'properties': {'a': {'type': 'string', 'title': 'A'}},
+                'required': ['a'],
+                'title': 'Pika',
+            },
+        },
     }
 
 
@@ -2713,21 +2707,18 @@ class Model(BaseModel):
 
         model_config = {'arbitrary_types_allowed': True}
 
+    # insert_assert(Model.model_json_schema())
     assert Model.model_json_schema() == {
-        'title': 'Model',
-        'type': 'object',
+        '$defs': {'CustomType': {'enum': ['a', 'b'], 'title': 'CustomType'}},
         'properties': {
             'data0': {
                 'anyOf': [{'type': 'string'}, {'items': {'type': 'string'}, 'type': 'array'}],
                 'title': 'Gen title',
             },
             'data1': {
-                'title': 'Data1 title',
+                'anyOf': [{'$ref': '#/$defs/CustomType'}, {'items': {'$ref': '#/$defs/CustomType'}, 'type': 'array'}],
                 'description': 'Data 1 description',
-                'anyOf': [
-                    {'$ref': '#/$defs/CustomType'},
-                    {'type': 'array', 'items': {'$ref': '#/$defs/CustomType'}},
-                ],
+                'title': 'Data1 title',
             },
             'data2': {
                 'description': 'Data 2',
@@ -2736,30 +2727,25 @@ class Model(BaseModel):
                 'title': 'Data2 title',
                 'type': 'array',
             },
-            'data3': {'title': 'Data3', 'type': 'array', 'items': {}},
+            'data3': {'items': {}, 'title': 'Data3', 'type': 'array'},
             'data4': {
+                'maxItems': 1,
+                'minItems': 1,
+                'prefixItems': [{'$ref': '#/$defs/CustomType'}],
                 'title': 'Data4',
                 'type': 'array',
-                'prefixItems': [{'$ref': '#/$defs/CustomType'}],
-                'minItems': 1,
-                'maxItems': 1,
             },
             'data5': {
+                'maxItems': 2,
+                'minItems': 2,
+                'prefixItems': [{'$ref': '#/$defs/CustomType'}, {'type': 'string'}],
                 'title': 'Data5',
                 'type': 'array',
-                'prefixItems': [{'$ref': '#/$defs/CustomType'}, {'type': 'string'}],
-                'minItems': 2,
-                'maxItems': 2,
             },
         },
         'required': ['data0', 'data1', 'data2', 'data3', 'data4', 'data5'],
-        '$defs': {
-            'CustomType': {
-                'title': 'CustomType title',
-                'enum': ['a', 'b'],
-                'type': 'string',
-            }
-        },
+        'title': 'Model',
+        'type': 'object',
     }
 
 
@@ -3484,6 +3470,7 @@ class Model(BaseModel):
     }
 
 
+@pytest.mark.xfail(reason='json schema generation for tagged unions is fundamentally incompatible with references')
 def test_alias_same():
     class Cat(BaseModel):
         pet_type: Literal['cat'] = Field(alias='typeOfPet')
@@ -3497,20 +3484,8 @@ class Model(BaseModel):
         pet: Union[Cat, Dog] = Field(discriminator='pet_type')
         number: int
 
+    # insert_assert(Model.model_json_schema())
     assert Model.model_json_schema() == {
-        'type': 'object',
-        'title': 'Model',
-        'properties': {
-            'number': {'title': 'Number', 'type': 'integer'},
-            'pet': {
-                'discriminator': {
-                    'mapping': {'cat': '#/$defs/Cat', 'dog': '#/$defs/Dog'},
-                    'propertyName': 'typeOfPet',
-                },
-                'oneOf': [{'$ref': '#/$defs/Cat'}, {'$ref': '#/$defs/Dog'}],
-                'title': 'Pet',
-            },
-        },
         '$defs': {
             'Cat': {
                 'properties': {
@@ -3531,7 +3506,17 @@ class Model(BaseModel):
                 'type': 'object',
             },
         },
+        'properties': {
+            'number': {'title': 'Number', 'type': 'integer'},
+            'pet': {
+                'oneOf': [{'$ref': '#/$defs/Cat'}, {'$ref': '#/$defs/Dog'}],
+                'title': 'Pet',
+                'discriminator': 'something',
+            },
+        },
         'required': ['pet', 'number'],
+        'title': 'Model',
+        'type': 'object',
     }
 
 
@@ -4451,3 +4436,83 @@ class Model(BaseModel):
         'title': 'Model',
         'type': 'object',
     }
+
+
+@pytest.mark.xfail(reason='Need to fix JSON schema customization for Enum')
+def test_override_enum_json_schema():
+    class CustomType(Enum):
+        A = 'a'
+        B = 'b'
+
+        @classmethod
+        def __get_pydantic_json_schema__(
+            cls, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler
+        ) -> core_schema.CoreSchema:
+            json_schema = handler(core_schema)
+            json_schema.update(title='CustomType title', type='string')
+            return json_schema
+
+    class Model(BaseModel):
+        x: CustomType
+
+    assert Model.model_json_schema() == {
+        'type': 'object',
+        'properties': {'x': {'$ref': '#/$defs/CustomType'}},
+        'required': ['x'],
+        'title': 'Model',
+        '$defs': {
+            'CustomType': {
+                'enum': ['a', 'b'],
+                'title': 'CustomType title',  # <----------- this is not getting set properly
+            }
+        },
+    }
+
+
+def test_json_schema_extras_on_ref() -> None:
+    @dataclass
+    class JsonSchemaExamples:
+        examples: Dict[str, Any]
+
+        def __get_pydantic_json_schema__(
+            self, core_schema: CoreSchema, handler: GetJsonSchemaHandler
+        ) -> JsonSchemaValue:
+            json_schema = handler(core_schema)
+            assert json_schema.keys() == {'$ref'}
+            json_schema['examples'] = to_json(self.examples)
+            return json_schema
+
+    @dataclass
+    class JsonSchemaTitle:
+        title: str
+
+        def __get_pydantic_json_schema__(
+            self, core_schema: CoreSchema, handler: GetJsonSchemaHandler
+        ) -> JsonSchemaValue:
+            json_schema = handler(core_schema)
+            assert json_schema.keys() == {'allOf', 'examples'}
+            json_schema['title'] = self.title
+            return json_schema
+
+    class Model(BaseModel):
+        name: str
+        age: int
+
+    ta = TypeAdapter(
+        Annotated[Model, JsonSchemaExamples({'foo': Model(name='John', age=28)}), JsonSchemaTitle('ModelTitle')]
+    )
+
+    # insert_assert(ta.json_schema())
+    assert ta.json_schema() == {
+        '$defs': {
+            'Model': {
+                'properties': {'age': {'title': 'Age', 'type': 'integer'}, 'name': {'title': 'Name', 'type': 'string'}},
+                'required': ['name', 'age'],
+                'title': 'Model',
+                'type': 'object',
+            }
+        },
+        'allOf': [{'$ref': '#/$defs/Model'}],
+        'examples': b'{"foo":{"name":"John","age":28}}',
+        'title': 'ModelTitle',
+    }
diff --git a/tests/test_main.py b/tests/test_main.py
index 1214ef981db..b33688dcb9c 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -2,6 +2,7 @@
 import platform
 import re
 import sys
+from collections import defaultdict
 from copy import deepcopy
 from dataclasses import dataclass
 from enum import Enum
@@ -22,7 +23,7 @@
 from uuid import UUID, uuid4
 
 import pytest
-from pydantic_core import CoreSchema
+from pydantic_core import CoreSchema, core_schema
 from typing_extensions import Annotated, Final, Literal
 
 from pydantic import (
@@ -39,6 +40,7 @@
     constr,
     field_validator,
 )
+from pydantic.type_adapter import TypeAdapter
 
 
 def test_success():
@@ -2361,3 +2363,84 @@ class Model(BaseModel):
     # we used to accidentally convert `__metadata__` to a list
     # which caused things like `typing.get_args()` to fail
     assert Model.__signature__.parameters['x'].annotation.__metadata__ == (123,)
+
+
+def test_get_core_schema_unpacks_refs_for_source_type() -> None:
+    # use a list to track since we end up calling `__get_pydantic_core_schema__` multiple times for models
+    # e.g. InnerModel.__get_pydantic_core_schema__ gets called:
+    # 1. When InnerModel is defined
+    # 2. When OuterModel is defined
+    # 3. When we use the TypeAdapter
+    received_schemas: dict[str, list[str]] = defaultdict(list)
+
+    @dataclass
+    class Marker:
+        name: str
+
+        def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema:
+            schema = handler(source_type)
+            received_schemas[self.name].append(schema['type'])
+            return schema
+
+    class InnerModel(BaseModel):
+        @classmethod
+        def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema:
+            schema = handler(source_type)
+            received_schemas['InnerModel'].append(schema['type'])
+            schema['metadata'] = schema.get('metadata', {})
+            schema['metadata']['foo'] = 'inner was here!'
+            return deepcopy(schema)
+
+    class OuterModel(BaseModel):
+        inner: Annotated[InnerModel, Marker('Marker("inner")')]
+
+        @classmethod
+        def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema:
+            schema = handler(source_type)
+            received_schemas['OuterModel'].append(schema['type'])
+            return schema
+
+    ta = TypeAdapter(Annotated[OuterModel, Marker('Marker("outer")')])
+
+    # super hacky check but it works in all cases and avoids a complex and fragile iteration over CoreSchema
+    # the point here is to verify that `__get_pydantic_core_schema__`
+    assert 'inner was here' in str(ta.core_schema)
+
+    assert received_schemas == {
+        'InnerModel': ['model', 'model', 'model'],
+        'Marker("inner")': ['definition-ref', 'definition-ref'],
+        'OuterModel': ['model', 'model'],
+        'Marker("outer")': ['definition-ref'],
+    }
+
+
+def test_get_core_schema_return_new_ref() -> None:
+    class InnerModel(BaseModel):
+        @classmethod
+        def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema:
+            schema = handler(source_type)
+            schema = deepcopy(schema)
+            schema['metadata'] = schema.get('metadata', {})
+            schema['metadata']['foo'] = 'inner was here!'
+            return deepcopy(schema)
+
+    class OuterModel(BaseModel):
+        inner: InnerModel
+        x: int = 1
+
+        @classmethod
+        def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema:
+            schema = handler(source_type)
+
+            def set_x(m: 'OuterModel') -> 'OuterModel':
+                m.x += 1
+                return m
+
+            return core_schema.no_info_after_validator_function(set_x, schema, ref=schema.pop('ref'))
+
+    cs = OuterModel.__pydantic_core_schema__
+    # super hacky check but it works in all cases and avoids a complex and fragile iteration over CoreSchema
+    # the point here is to verify that `__get_pydantic_core_schema__`
+    assert 'inner was here' in str(cs)
+
+    assert OuterModel(inner=InnerModel()).x == 2
diff --git a/tests/test_parse.py b/tests/test_parse.py
index 9cf95783e98..dc6e356bed1 100644
--- a/tests/test_parse.py
+++ b/tests/test_parse.py
@@ -1,9 +1,11 @@
 from typing import List, Tuple
 
 import pytest
+from pydantic_core import CoreSchema
 
-from pydantic import BaseModel, ValidationError, model_validator, parse_obj_as
+from pydantic import BaseModel, GetJsonSchemaHandler, ValidationError, model_validator, parse_obj_as
 from pydantic.functional_serializers import model_serializer
+from pydantic.json_schema import JsonSchemaValue
 
 
 class Model(BaseModel):
@@ -76,9 +78,12 @@ def _serialize(self, handler, info):
                 return data
 
         @classmethod
-        def __get_pydantic_json_schema__(cls, core_schema, handler):
+        def __get_pydantic_json_schema__(
+            cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler
+        ) -> JsonSchemaValue:
             json_schema = handler(core_schema)
-            return json_schema['properties']['root']
+            root = handler.resolve_ref_schema(json_schema)['properties']['root']
+            return root
 
     # Validation
     m = MyModel.model_validate('a')
diff --git a/tests/test_root_model.py b/tests/test_root_model.py
index 743ea223ef5..709c00845c4 100644
--- a/tests/test_root_model.py
+++ b/tests/test_root_model.py
@@ -1,6 +1,7 @@
 from typing import Any, Dict, List, Optional
 
 import pytest
+from pydantic_core import CoreSchema
 from pydantic_core.core_schema import SerializerFunctionWrapHandler
 
 from pydantic import (
@@ -39,13 +40,25 @@ class InnerModel(BaseModel):
     )
 
 
+def check_schema(schema: CoreSchema) -> None:
+    # we assume the shape of the core schema here, which is not a guarantee
+    # pydantic makes to it's users but is useful to check here to make sure
+    # we are doing the right thing internally
+    assert schema['type'] == 'definitions'
+    inner = schema['schema']
+    assert inner['type'] == 'definition-ref'
+    ref = inner['schema_ref']  # type: ignore
+    schema = next(s for s in schema['definitions'] if s['ref'] == ref)  # type: ignore
+    assert schema['type'] == 'model'
+    assert schema['root_model'] is True
+    assert schema['custom_init'] is False
+
+
 @parametrize_root_model()
 def test_root_model_specialized(root_type, root_value, dump_value):
     Model = RootModel[root_type]
 
-    assert Model.__pydantic_core_schema__['type'] == 'model'
-    assert Model.__pydantic_core_schema__['root_model'] is True
-    assert Model.__pydantic_core_schema__['custom_init'] is False
+    check_schema(Model.__pydantic_core_schema__)
 
     m = Model(root_value)
 
@@ -58,9 +71,7 @@ def test_root_model_inherited(root_type, root_value, dump_value):
     class Model(RootModel[root_type]):
         pass
 
-    assert Model.__pydantic_core_schema__['type'] == 'model'
-    assert Model.__pydantic_core_schema__['root_model'] is True
-    assert Model.__pydantic_core_schema__['custom_init'] is False
+    check_schema(Model.__pydantic_core_schema__)
 
     m = Model(root_value)
 
diff --git a/tests/test_type_alias_type.py b/tests/test_type_alias_type.py
index bfccb817b29..ffd84b3fa61 100644
--- a/tests/test_type_alias_type.py
+++ b/tests/test_type_alias_type.py
@@ -153,7 +153,6 @@ def test_type_alias_annotated() -> None:
     assert t.json_schema() == {'type': 'array', 'items': {'type': 'integer'}, 'maxItems': 1}
 
 
-@pytest.mark.xfail(reason='Not working yet, we cannot apply constraints to named/ref types')
 def test_type_alias_annotated_defs() -> None:
     # force use of refs by referencing the schema in multiple places
     t = TypeAdapter(Tuple[ShortMyList[int], ShortMyList[int]])
@@ -231,7 +230,6 @@ def test_recursive_generic_type_alias() -> None:
     }
 
 
-@pytest.mark.xfail(reason='Not working yet, we cannot apply constraints to named/ref types')
 def test_recursive_generic_type_alias_annotated() -> None:
     t = TypeAdapter(ShortRecursiveGenericAlias[int])
 
@@ -249,6 +247,7 @@ def test_recursive_generic_type_alias_annotated() -> None:
         }
     ]
 
+    # insert_assert(t.json_schema())
     assert t.json_schema() == {
         'type': 'array',
         'items': {'anyOf': [{'$ref': '#/$defs/RecursiveGenericAlias'}, {'type': 'integer'}]},
@@ -262,7 +261,6 @@ def test_recursive_generic_type_alias_annotated() -> None:
     }
 
 
-@pytest.mark.xfail(reason='Not working yet, we cannot apply constraints to named/ref types')
 def test_recursive_generic_type_alias_annotated_defs() -> None:
     # force use of refs by referencing the schema in multiple places
     t = TypeAdapter(Tuple[ShortRecursiveGenericAlias[int], ShortRecursiveGenericAlias[int]])
@@ -281,6 +279,7 @@ def test_recursive_generic_type_alias_annotated_defs() -> None:
         }
     ]
 
+    # insert_assert(t.json_schema())
     assert t.json_schema() == {
         'type': 'array',
         'minItems': 2,
diff --git a/tests/test_types_typeddict.py b/tests/test_types_typeddict.py
index ba33d77a373..75956cc0620 100644
--- a/tests/test_types_typeddict.py
+++ b/tests/test_types_typeddict.py
@@ -201,9 +201,11 @@ def __get_pydantic_core_schema__(
             cls, source_type: Any, handler: GetCoreSchemaHandler
         ) -> core_schema.CoreSchema:
             schema = handler(source_type)
-            schema['computed_fields'] = [
-                core_schema.computed_field(property_name='another_b', return_schema=core_schema.int_schema())
-            ]
+            schema = handler.resolve_ref_schema(schema)
+            assert schema['type'] == 'typed-dict'
+            b = schema['fields']['b']['schema']
+            assert b['type'] == 'int'
+            b['gt'] = 0  # type: ignore
             return schema
 
     class Model(BaseModel):
@@ -211,66 +213,66 @@ class Model(BaseModel):
         data_td: DataTD
         custom_td: CustomTD
 
+    # insert_assert(Model.model_json_schema(mode='validation'))
     assert Model.model_json_schema(mode='validation') == {
-        'title': 'Model',
         'type': 'object',
         'properties': {
-            'custom_td': {'$ref': '#/$defs/CustomTD'},
             'data': {'$ref': '#/$defs/Data'},
             'data_td': {'$ref': '#/$defs/DataTD'},
+            'custom_td': {'$ref': '#/$defs/CustomTD'},
         },
         'required': ['data', 'data_td', 'custom_td'],
+        'title': 'Model',
         '$defs': {
+            'DataTD': {
+                'type': 'object',
+                'properties': {'a': {'type': 'integer', 'title': 'A'}},
+                'required': ['a'],
+                'title': 'DataTD',
+            },
             'CustomTD': {
                 'type': 'object',
-                'title': 'CustomTD',
-                'properties': {'b': {'title': 'B', 'type': 'integer'}},
+                'properties': {'b': {'type': 'integer', 'exclusiveMinimum': 0, 'title': 'B'}},
                 'required': ['b'],
+                'title': 'CustomTD',
             },
             'Data': {
                 'type': 'object',
-                'title': 'Data',
-                'properties': {'a': {'title': 'A', 'type': 'integer'}},
-                'required': ['a'],
-            },
-            'DataTD': {
-                'type': 'object',
-                'title': 'DataTD',
-                'properties': {'a': {'title': 'A', 'type': 'integer'}},
+                'properties': {'a': {'type': 'integer', 'title': 'A'}},
                 'required': ['a'],
+                'title': 'Data',
             },
         },
     }
+
+    # insert_assert(Model.model_json_schema(mode='serialization'))
     assert Model.model_json_schema(mode='serialization') == {
-        'title': 'Model',
         'type': 'object',
         'properties': {
-            'custom_td': {'$ref': '#/$defs/CustomTD'},
             'data': {'$ref': '#/$defs/Data'},
             'data_td': {'$ref': '#/$defs/DataTD'},
+            'custom_td': {'$ref': '#/$defs/CustomTD'},
         },
         'required': ['data', 'data_td', 'custom_td'],
+        'title': 'Model',
         '$defs': {
+            'DataTD': {
+                'type': 'object',
+                'properties': {'a': {'type': 'integer', 'title': 'A'}},
+                'required': ['a'],
+                'title': 'DataTD',
+            },
             'CustomTD': {
                 'type': 'object',
+                'properties': {'b': {'type': 'integer', 'exclusiveMinimum': 0, 'title': 'B'}},
+                'required': ['b'],
                 'title': 'CustomTD',
-                'properties': {
-                    'b': {'title': 'B', 'type': 'integer'},
-                    'another_b': {'title': 'Another B', 'type': 'integer'},
-                },
-                'required': ['b', 'another_b'],
             },
             'Data': {
                 'type': 'object',
-                'title': 'Data',
-                'properties': {'a': {'title': 'A', 'type': 'integer'}},
-                'required': ['a'],
-            },
-            'DataTD': {
-                'type': 'object',
-                'title': 'DataTD',
-                'properties': {'a': {'title': 'A', 'type': 'integer'}},
+                'properties': {'a': {'type': 'integer', 'title': 'A'}},
                 'required': ['a'],
+                'title': 'Data',
             },
         },
     }


Code review comments:
diff_hunk: @@ -1504,3 +1477,50 @@ def _common_field(
         'frozen': frozen,
         'metadata': metadata,
     }
+
+
+class _Definitions:
+    """Keeps track of references and definitions"""
+
+    def __init__(self) -> None:
+        self.seen: set[str] = set()
+        self.definitions: dict[str, core_schema.CoreSchema] = {}
+
+    @contextmanager
+    def get_schema_or_ref(self, tp: Any) -> Iterator[tuple[str, None] | tuple[str, CoreSchema]]:
+        """
+        Get a definition for `tp` if one exists.
+
+        If a definition exists a tuple of (ref_string, CoreSchema) is returned.
+        If no definition exists yet a tuple of (ref_string, None) is returned.
path: pydantic/_internal/_generate_schema.py
text: @user1:
```suggestion
        If no definition exists yet, a tuple of `(ref_string, None)` is returned.
```

diff_hunk: @@ -1504,3 +1477,50 @@ def _common_field(
         'frozen': frozen,
         'metadata': metadata,
     }
+
+
+class _Definitions:
+    """Keeps track of references and definitions"""
+
+    def __init__(self) -> None:
+        self.seen: set[str] = set()
+        self.definitions: dict[str, core_schema.CoreSchema] = {}
+
+    @contextmanager
+    def get_schema_or_ref(self, tp: Any) -> Iterator[tuple[str, None] | tuple[str, CoreSchema]]:
+        """
+        Get a definition for `tp` if one exists.
+
+        If a definition exists a tuple of (ref_string, CoreSchema) is returned.
path: pydantic/_internal/_generate_schema.py
text: @user1:
```suggestion
        If a definition exists, a tuple of `(ref_string, CoreSchema)` is returned.
```