diff --git a/elasticgraph-apollo/apollo_tests_implementation/config.ru b/elasticgraph-apollo/apollo_tests_implementation/config.ru
index a2fe8b65e..feaa7b436 100644
--- a/elasticgraph-apollo/apollo_tests_implementation/config.ru
+++ b/elasticgraph-apollo/apollo_tests_implementation/config.ru
@@ -107,7 +107,7 @@ events = records_by_type.flat_map do |type_name, records|
     {
       __typename: type_name,
       __version: 1,
-      __json_schema_version: 1
+      __schema_version: 1
     }.merge(record)
   end
 
diff --git a/elasticgraph-datastore_core/spec/integration/elastic_graph/datastore_core/index_definition/rollover_index_template_spec.rb b/elasticgraph-datastore_core/spec/integration/elastic_graph/datastore_core/index_definition/rollover_index_template_spec.rb
index f7d05161b..6a0bbd515 100644
--- a/elasticgraph-datastore_core/spec/integration/elastic_graph/datastore_core/index_definition/rollover_index_template_spec.rb
+++ b/elasticgraph-datastore_core/spec/integration/elastic_graph/datastore_core/index_definition/rollover_index_template_spec.rb
@@ -54,7 +54,7 @@ def configure_index(index)
               "created_at" => "2019-06-02T12:00:00Z",
               "__typename" => "Widget",
               "__version" => 1,
-              "__json_schema_version" => 1
+              "__schema_version" => 1
             }
             index_name_for_writes = index_definition.index_name_for_writes(record)
             derive_index_from_template(record, datastore_core)
diff --git a/elasticgraph-graphql/spec/acceptance/elasticgraph_graphql_acceptance_support.rb b/elasticgraph-graphql/spec/acceptance/elasticgraph_graphql_acceptance_support.rb
index 1d5d0b4e6..401c0211b 100644
--- a/elasticgraph-graphql/spec/acceptance/elasticgraph_graphql_acceptance_support.rb
+++ b/elasticgraph-graphql/spec/acceptance/elasticgraph_graphql_acceptance_support.rb
@@ -315,7 +315,7 @@ def update_enum_values_in(data, json_schema_defs, type_name)
         else
           props = json_schema_def.fetch("properties")
           data.to_h do |field_name, field_value|
-            unless [:__version, :__typename, :__json_schema_version].include?(field_name)
+            unless [:__version, :__typename, :__schema_version].include?(field_name)
               field_type = props.fetch(word_to_snake_case(field_name.to_s)).fetch("ElasticGraph").fetch("type")[/\w+/]
               field_value = update_enum_values_in(field_value, json_schema_defs, field_type)
             end
diff --git a/elasticgraph-indexer/README.md b/elasticgraph-indexer/README.md
index 407441965..95b1f6b70 100644
--- a/elasticgraph-indexer/README.md
+++ b/elasticgraph-indexer/README.md
@@ -89,3 +89,23 @@ module MyCompany
   end
 end
 ```
+
+A decoded event hash may carry a `schema_version` to request a specific schema artifact version. The
+key is optional, because an ingestion format may have no versions at all. Each ingestion adapter
+decides what a missing version means for its own format. `elasticgraph-json_ingestion` uses the latest
+available JSON schema version.
+
+Decoders identify their format with an `ingestion_format` key (for example, `"json"`). Adapters use
+that tag to route events independently of whether the format has schema versions. Untagged events
+remain compatible with JSON callers; other formats must supply a tag. A sole adapter receives all
+untagged events so it can provide detailed validation errors, but must still recognize an explicit tag.
+
+A successful ingestion adapter result supplies both a record preparer and a normalized event through
+`IngestionAdapter::ValidationResult.valid(record_preparer, event: normalized_event)`. The normalized
+event must preserve the event's identity, record, and transport metadata without mutating the input.
+For versioned formats, set `schema_version` to the artifact version actually selected for validation
+and preparation. Formats without versions may omit it. Operations, latency logs, and warehouse
+partitions use this normalized envelope.
+
+See [the JSON ingestion upgrade guide](../elasticgraph-json_ingestion/README.md#upgrading-an-existing-json-deployment)
+when upgrading an existing deployment.
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb
index 480970121..75cbc1a15 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb
@@ -47,11 +47,11 @@ class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms
           },
           skip_record_validation_percents_by_type: {
             description: "Map of GraphQL type names to the percentage of records of that type whose per-record " \
-              "JSON schema validation should be skipped. `0` (or an absent key) validates every record of the " \
+              "schema validation should be skipped. `0` (or an absent key) validates every record of the " \
               "type; `100` skips every record; values in between sample, and may be fractional. The decision is " \
               "deterministic per event id (`type:id@vversion`), so the same event makes the same choice on every " \
-              "retry and on every indexer pod. The event envelope (op, id, type, version, json_schema_version, " \
-              "latency_timestamps) is always validated, regardless of this setting.\n\n" \
+              "retry and on every indexer pod. The ingestion adapter always validates the event envelope, " \
+              "including any format-specific schema version, regardless of this setting.\n\n" \
               "With a large schema the per-record schema walk consumes a significant share of indexing CPU: every " \
               "record is checked against every regex, enum, min/max, format, and abstract-type discriminator " \
               "defined for its type. Skipping it trades that check for throughput, which is worthwhile when " \
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb
index 4f0bbb018..647e9c91d 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb
@@ -21,8 +21,13 @@ def initialize(config:, schema_artifacts:, logger:)
           # must be defined, but nothing to do
         end
 
+        # Tag each decoded event with `ingestion_format` so that adapters can recognize their format.
+        # The tag is independent of schema version; untagged events default to JSON when JSON is available.
+        #
         # @param payload [String] a raw payload from the transport
-        # @return [Array<Hash<String, Object>>] the decoded ElasticGraph indexing events
+        # @return [Array<Hash<String, Object>>] the decoded ElasticGraph indexing events. An event may
+        #   include a `schema_version`, but does not have to: an ingestion format with no versions
+        #   omits it. Each ingestion adapter decides what a missing version means for its own format.
         def decode(payload)
           # :nocov: -- must return an array to satisfy Steep type checking but never called
           []
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb
index 5e5975a6d..bf0d3c435 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb
@@ -23,7 +23,7 @@ def initialize(schema_artifacts:, logger:)
 
         # Indicates whether this adapter recognizes the given event as one of its own. When multiple
         # adapters are available, the indexer routes each event to the first adapter that returns
-        # `true`. (When exactly one adapter is available, it receives all events.)
+        # `true`. (When exactly one adapter is available, it receives all untagged events.)
         #
         # @param event [Hash<String, Object>] an ElasticGraph indexing event
         # @return [Boolean] whether this adapter handles the event
@@ -34,14 +34,19 @@ def handles_event?(event)
         end
 
         # Validates the given event and resolves the record preparer appropriate for the event's
-        # schema version.
+        # schema version. The successful result carries a normalized copy of the event, including
+        # the selected `schema_version` when the format is versioned. Preserve event identity,
+        # record data, and transport metadata; do not mutate the caller's event.
+        #
+        # The event's `schema_version` is optional, because an ingestion format may have no versions
+        # at all. Each adapter decides what a missing version means for its own format.
         #
         # @param event [Hash<String, Object>] an ElasticGraph indexing event
         # @param skip_record_validation [Boolean] whether to skip record validation; the event envelope must still be validated
         # @return [ValidationResult] the result of validating the event
         def validate_event(event, skip_record_validation: false)
           # simplecov:disable -- must return a result to satisfy Steep type checking but never called
-          ValidationResult.valid(RecordPreparer::Identity)
+          ValidationResult.valid(RecordPreparer::Identity, event: event)
           # simplecov:enable
         end
       end
@@ -55,22 +60,25 @@ def validate_event(event, skip_record_validation: false)
       Failure = ::Data.define(:payload_description, :message)
 
       # Returned by {Interface#validate_event}. Either `failure` is non-nil (the event was invalid)
-      # or `record_preparer` is non-nil (the event was valid and its record can be prepared for
+      # or `event` and `record_preparer` are non-nil (the event was valid and its record can be prepared for
       # indexing with the given preparer).
       #
+      # @!attribute [r] event
+      #   @return [Hash<String, Object>, nil] normalized event, when validation succeeds
       # @!attribute [r] record_preparer
       #   @return [Object, nil] preparer for the event's record, when the event is valid
       # @!attribute [r] failure
       #   @return [Failure, nil] description of the validation problem, when the event is invalid
-      ValidationResult = ::Data.define(:record_preparer, :failure) do
+      ValidationResult = ::Data.define(:event, :record_preparer, :failure) do
         # @implements ValidationResult
 
         # Builds a result for a valid event.
         #
+        # @param event [Hash<String, Object>] normalized event
         # @param record_preparer [Object] preparer for the event's record
         # @return [ValidationResult]
-        def self.valid(record_preparer)
-          new(record_preparer: record_preparer, failure: nil)
+        def self.valid(record_preparer, event:)
+          new(event: event, record_preparer: record_preparer, failure: nil)
         end
 
         # Builds a result for an invalid event.
@@ -79,7 +87,7 @@ def self.valid(record_preparer)
         # @param message [String] detailed validation failure message
         # @return [ValidationResult]
         def self.invalid(payload_description:, message:)
-          new(record_preparer: nil, failure: Failure.new(payload_description: payload_description, message: message))
+          new(event: nil, record_preparer: nil, failure: Failure.new(payload_description: payload_description, message: message))
         end
       end
     end
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
index 755aeb457..6dc9fdcfe 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
@@ -6,6 +6,7 @@
 #
 # frozen_string_literal: true
 
+require "elastic_graph/constants"
 require "elastic_graph/indexer/event_id"
 require "elastic_graph/indexer/failed_event_error"
 require "elastic_graph/indexer/operation/update"
@@ -38,6 +39,7 @@ def build(event)
             return build_failed_result(event, failure.payload_description, failure.message)
           end
 
+          event = validation_result.event # : event
           record_preparer = validation_result.record_preparer # : _RecordPreparer
           if skip_record_validation
             build_success_result_isolating_malformed_records(event, record_preparer, adapter)
@@ -75,10 +77,10 @@ def build_success_result_isolating_malformed_records(event, record_preparer, ada
         end
 
         # Routes the event to the first ingestion adapter that recognizes it. When exactly one
-        # adapter is available, it receives all events--including unrecognizable ones--so that
+        # adapter is available, it receives all untagged events--including unrecognizable ones--so that
         # its more specific validation failure messages are used.
         def ingestion_adapter_for(event)
-          return ingestion_adapters.first if ingestion_adapters.one?
+          return ingestion_adapters.first if ingestion_adapters.one? && !event.key?(INGESTION_FORMAT_KEY)
 
           ingestion_adapters.find { |adapter| adapter.handles_event?(event) }
         end
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/processor.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/processor.rb
index 23cfe0c3e..8fe9c4e78 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/processor.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/processor.rb
@@ -41,21 +41,24 @@ def process(events, refresh_indices: false)
       # Like `process`, but returns failures instead of raising an exception.
       # The caller is responsible for handling the failures.
       def process_returning_failures(events, refresh_indices: false)
-        factory_results_by_event = events.to_h { |event| [event, @operation_factory.build(event)] }
-
-        factory_results = factory_results_by_event.values
+        factory_results = events.uniq.map { |event| @operation_factory.build(event) }
+        operations = factory_results.flat_map(&:operations)
 
         log_skipped_record_validations(factory_results)
 
-        bulk_result = @datastore_router.bulk(factory_results.flat_map(&:operations), refresh: refresh_indices)
+        bulk_result = @datastore_router.bulk(operations, refresh: refresh_indices)
         successful_operations = bulk_result.successful_operations(check_failures: false)
 
         calculate_latency_metrics(successful_operations, bulk_result.noop_results)
 
+        # Adapters may normalize event envelopes, so correlate failures using the events on the
+        # operations rather than the original input hashes.
+        operations_by_event = operations.group_by(&:event)
+
         all_failures =
           factory_results.map(&:failed_event_error).compact +
           bulk_result.failure_results.map do |result|
-            all_operations_for_event = factory_results_by_event.fetch(result.event).operations
+            all_operations_for_event = operations_by_event.fetch(result.event)
             FailedEventError.from_failed_operation_result(result, all_operations_for_event.to_set)
           end
 
@@ -142,12 +145,18 @@ def calculate_latency_metrics(successful_operations, noop_results)
 
           result = successful_events.include?(event) ? "success" : "noop"
 
+          # The schema version is optional, since an ingestion format may have no versions at all.
+          schema_version = event[SCHEMA_VERSION_KEY]
+
           @logger.info({
             "message_type" => "ElasticGraphIndexingLatencies",
             "message_id" => event["message_id"],
             "event_type" => event.fetch("type"),
             "event_id" => EventID.from_event(event).to_s,
-            JSON_SCHEMA_VERSION_KEY => event.fetch(JSON_SCHEMA_VERSION_KEY),
+            SCHEMA_VERSION_KEY => schema_version,
+            # Deprecated alias of `schema_version`, kept so that dashboards and monitors that watch
+            # the old name keep working.
+            JSON_SCHEMA_VERSION_KEY => schema_version,
             "latencies_in_ms_from" => latencies_in_ms_from,
             "slo_results" => slo_results,
             "result" => result
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/test_support/converters.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/test_support/converters.rb
index 31de51c26..afd2cf24c 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/test_support/converters.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/test_support/converters.rb
@@ -14,16 +14,28 @@ module ElasticGraph
   class Indexer
     module TestSupport
       module Converters
+        # Attributes that describe the event rather than the record, so they never reach the record.
+        # `__json_schema_version` is the legacy name of `__schema_version`; projects generated before
+        # the schema version became ingestion-format-neutral still use it.
+        EVENT_ONLY_ATTRIBUTES = ["__typename", "__version", "__schema_version", "__json_schema_version"]
+
         # Helper method for testing and generating fake data to convert a factory record into an event
         def self.upsert_event_for(record)
-          {
+          event = {
             "op" => "upsert",
             "id" => record.fetch("id"),
             "type" => record.fetch("__typename"),
             "version" => record.fetch("__version"),
-            "record" => record.except("__typename", "__version", "__json_schema_version"),
-            JSON_SCHEMA_VERSION_KEY => record.fetch("__json_schema_version")
+            "record" => record.except(*EVENT_ONLY_ATTRIBUTES)
           }
+
+          # The schema version is optional, so include it only when the factory supplies one.
+          schema_version = record.fetch("__schema_version") { record["__json_schema_version"] }
+          unless schema_version.nil?
+            event[SCHEMA_VERSION_KEY] = schema_version
+          end
+
+          event
         end
 
         # Helper method to create an array of events given an array of records
diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs
index 47a17ee56..2363828e3 100644
--- a/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs
+++ b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs
@@ -25,16 +25,17 @@ module ElasticGraph
       end
 
       class ValidationResultSupertype < ::Data
+        attr_reader event: event?
         attr_reader record_preparer: _RecordPreparer?
         attr_reader failure: Failure?
 
-        def initialize: (record_preparer: _RecordPreparer?, failure: Failure?) -> void
-        def self.new: (record_preparer: _RecordPreparer?, failure: Failure?) -> instance
+        def initialize: (event: event?, record_preparer: _RecordPreparer?, failure: Failure?) -> void
+        def self.new: (event: event?, record_preparer: _RecordPreparer?, failure: Failure?) -> instance
         def self.members: () -> ::Array[::Symbol]
       end
 
       class ValidationResult < ValidationResultSupertype
-        def self.valid: (_RecordPreparer) -> ValidationResult
+        def self.valid: (_RecordPreparer, event: event) -> ValidationResult
         def self.invalid: (payload_description: ::String, message: ::String) -> ValidationResult
       end
     end
diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/test_support/converters.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/test_support/converters.rbs
index 61ff910e3..f595f5160 100644
--- a/elasticgraph-indexer/sig/elastic_graph/indexer/test_support/converters.rbs
+++ b/elasticgraph-indexer/sig/elastic_graph/indexer/test_support/converters.rbs
@@ -2,6 +2,8 @@ module ElasticGraph
   class Indexer
     module TestSupport
       module Converters
+        EVENT_ONLY_ATTRIBUTES: ::Array[::String]
+
         def self.upsert_event_for: (::Hash[::String, untyped]) -> ::Hash[::String, untyped]
 
         def self.upsert_events_for_records: (
diff --git a/elasticgraph-indexer/spec/acceptance/schema_evolution_spec.rb b/elasticgraph-indexer/spec/acceptance/schema_evolution_spec.rb
index 314dc4eea..559cca462 100644
--- a/elasticgraph-indexer/spec/acceptance/schema_evolution_spec.rb
+++ b/elasticgraph-indexer/spec/acceptance/schema_evolution_spec.rb
@@ -96,7 +96,7 @@ def build_address_event_without_geolocation
       end
 
       def build_widget(json_schema_version:)
-        event = build_upsert_event(:widget, __json_schema_version: json_schema_version)
+        event = build_upsert_event(:widget, __schema_version: json_schema_version)
         event.merge("record" => (yield event.fetch("record")))
       end
     end
@@ -116,7 +116,7 @@ def build_widget(json_schema_version:)
         write_address_schema_def(json_schema_version: 2, address_extras: "t.deleted_field 'deprecated'")
         dump_artifacts
 
-        event = build_upsert_event(:address, id: "abc", deprecated: "foo", __json_schema_version: 1)
+        event = build_upsert_event(:address, id: "abc", deprecated: "foo", __schema_version: 1)
         expect(event.dig("record", "deprecated")).to eq("foo")
 
         boot_indexer.processor.process([event], refresh_indices: true)
@@ -162,8 +162,8 @@ def get_address_payload(id)
         # included at that part of the JSON schema. So here we verify that the factory includes that.
         expect(build(:team_season)).to include(__typename: "TeamSeason")
 
-        v1_event = build_upsert_event(:team, __json_schema_version: 1)
-        v2_event = build_upsert_event(:team, __json_schema_version: 2)
+        v1_event = build_upsert_event(:team, __schema_version: 1)
+        v2_event = build_upsert_event(:team, __schema_version: 2)
           .then { |event| ::JSON.generate(event) }
           # Fix the event to align with the v2 schema, since `build_upsert_event` doesn't automatically
           # know that the `__typename` should be `SeasonOfATeam` instead of `TeamSeason`.
@@ -200,8 +200,8 @@ def get_address_payload(id)
           end
           dump_artifacts
 
-          v1_event = build_upsert_event(:team, __json_schema_version: 1)
-          v2_event = build_upsert_event(:team, __json_schema_version: 2)
+          v1_event = build_upsert_event(:team, __schema_version: 1)
+          v2_event = build_upsert_event(:team, __schema_version: 2)
 
           expect {
             boot_indexer.processor.process([v1_event, v2_event], refresh_indices: true)
@@ -244,9 +244,9 @@ def get_address_payload(id)
           end
           dump_artifacts
 
-          v1_event = build_upsert_event(:team, __json_schema_version: 1)
+          v1_event = build_upsert_event(:team, __schema_version: 1)
           v1_event = ::JSON.parse(::JSON.generate(v1_event).gsub('"name":', '"full_name":'))
-          v2_event = build_upsert_event(:team, __json_schema_version: 2)
+          v2_event = build_upsert_event(:team, __schema_version: 2)
 
           expect {
             boot_indexer.processor.process([v1_event, v2_event], refresh_indices: true)
@@ -288,7 +288,7 @@ def get_address_payload(id)
           end
           dump_artifacts
 
-          v1_event = build_upsert_event(:team, __json_schema_version: 1)
+          v1_event = build_upsert_event(:team, __schema_version: 1)
 
           expect {
             boot_indexer.processor.process([v1_event], refresh_indices: true)
@@ -323,7 +323,7 @@ def get_address_payload(id)
           write_address_schema_def(json_schema_version: 2, schema_extras: 'schema.deleted_type "Team"')
           dump_artifacts
 
-          v1_event = build_upsert_event(:team, __json_schema_version: 1)
+          v1_event = build_upsert_event(:team, __schema_version: 1)
           boot_indexer.processor.process([v1_event], refresh_indices: true)
 
           expect(search_for_ids("teams")).to be_empty
diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb
index 4301e4ca6..8e1847213 100644
--- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb
+++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb
@@ -9,6 +9,7 @@
 require "elastic_graph/constants"
 require "elastic_graph/indexer"
 require "elastic_graph/indexer/operation/factory"
+require "elastic_graph/json_ingestion/indexing_event_decoder"
 require "elastic_graph/json_ingestion/record_preparer_factory"
 require "elastic_graph/spec_support/builds_indexer_operation"
 require "json"
@@ -37,7 +38,7 @@ module Operation
               "type" => "Widget",
               "version" => 1,
               "record" => event["record"],
-              JSON_SCHEMA_VERSION_KEY => 1
+              SCHEMA_VERSION_KEY => 1
             }
 
             expect(build_expecting_success(event)).to contain_exactly(
@@ -103,7 +104,7 @@ module Operation
                 "type" => "Component",
                 "version" => 1,
                 "record" => event["record"],
-                JSON_SCHEMA_VERSION_KEY => 1
+                SCHEMA_VERSION_KEY => 1
               })])
             end
 
@@ -212,7 +213,7 @@ module Operation
                 "type" => "Widget",
                 "version" => 1,
                 "record" => event["record"],
-                JSON_SCHEMA_VERSION_KEY => 1
+                SCHEMA_VERSION_KEY => 1
               }
 
               expect(build_expecting_success(event)).to contain_exactly(
@@ -335,7 +336,7 @@ def factory_whose_record_preparation_is_broken
               "type" => "Component",
               "version" => 1,
               "record" => event["record"],
-              JSON_SCHEMA_VERSION_KEY => 1
+              SCHEMA_VERSION_KEY => 1
             }.merge(latency_timestamps))])
           end
 
@@ -345,7 +346,7 @@ def factory_whose_record_preparation_is_broken
               "id" => "1",
               "type" => "MyOwnInvalidGraphQlType",
               "version" => 1,
-              JSON_SCHEMA_VERSION_KEY => 1,
+              SCHEMA_VERSION_KEY => 1,
               "record" => {"field1" => "value1", "field2" => "value2", "id" => "1"}
             }
 
@@ -359,7 +360,7 @@ def factory_whose_record_preparation_is_broken
               "id" => "1",
               "type" => "WidgetOptions",
               "version" => 1,
-              JSON_SCHEMA_VERSION_KEY => 1,
+              SCHEMA_VERSION_KEY => 1,
               "record" => {"field1" => "value1", "field2" => "value2", "id" => "1"}
             }
 
@@ -376,17 +377,17 @@ def factory_whose_record_preparation_is_broken
             expect_failed_event_error(event, "missing_keys", "type", expect_no_ops: true)
           end
 
-          it "notifies an error on missing `#{JSON_SCHEMA_VERSION_KEY}`" do
-            event = build_upsert_event(:component).except(JSON_SCHEMA_VERSION_KEY)
+          it "builds operations for an event that carries no `#{SCHEMA_VERSION_KEY}`, since the key is optional" do
+            event = build_upsert_event(:component).except(SCHEMA_VERSION_KEY)
 
-            expect_failed_event_error(event, JSON_SCHEMA_VERSION_KEY)
+            expect(build_expecting_success(event)).not_to be_empty
           end
 
           it "notifies an error on wrong field types" do
             event = {
               "op" => "upsert",
               "id" => 1,
-              JSON_SCHEMA_VERSION_KEY => 1,
+              SCHEMA_VERSION_KEY => 1,
               "type" => [],
               "version" => "1",
               "record" => ""
@@ -471,6 +472,44 @@ def factory_whose_record_preparation_is_broken
           end
 
           context "when multiple ingestion adapters are available" do
+            it "routes tagged formats independently of adapter order and schema version" do
+              other_adapter = Class.new do
+                def handles_event?(event)
+                  event[INGESTION_FORMAT_KEY] == "other"
+                end
+
+                def validate_event(event, skip_record_validation: false)
+                  IngestionAdapter::ValidationResult.valid(RecordPreparer::Identity, event: event)
+                end
+              end.new
+              json_adapter = indexer.ingestion_adapters.first
+              allow(other_adapter).to receive(:validate_event).and_call_original
+              decoder = JSONIngestion::IndexingEventDecoder.new(config: {}, schema_artifacts: indexer.schema_artifacts, logger: indexer.logger)
+              versionless_event = build_upsert_event(:component).except(SCHEMA_VERSION_KEY)
+              json_events = [versionless_event, decoder.decode(::JSON.generate(versionless_event)).first]
+              other_events = [versionless_event, versionless_event.merge(SCHEMA_VERSION_KEY => 7)].map do |event|
+                event.merge(INGESTION_FORMAT_KEY => "other")
+              end
+
+              [json_adapter, other_adapter].permutation.each do |adapters|
+                factory = indexer.operation_factory.with(ingestion_adapters: adapters)
+
+                json_events.each do |event|
+                  result = factory.build(event)
+                  expect(result.failed_event_error).to be nil
+                  expect(result.operations.first.event.fetch(SCHEMA_VERSION_KEY)).to eq(1)
+                end
+
+                other_events.each do |event|
+                  result = factory.build(event)
+                  expect(result.failed_event_error).to be nil
+                  expect(result.operations.first.event).to eq(event)
+                end
+              end
+
+              expect(other_adapter).to have_received(:validate_event).exactly(4).times
+            end
+
             it "routes each event to the first adapter that recognizes it" do
               event = build_upsert_event(:component, id: "1", __version: 1)
 
@@ -501,6 +540,12 @@ def factory_whose_record_preparation_is_broken
           end
 
           context "when a single ingestion adapter is available" do
+            it "rejects an explicitly different format instead of sending it to the sole adapter" do
+              event = build_upsert_event(:component).merge(INGESTION_FORMAT_KEY => "other")
+
+              expect_failed_event_error(event, "No available ingestion adapter recognized this event.")
+            end
+
             it "routes all events to it, even ones it does not recognize, so that its more specific failure messages are used" do
               event = build_upsert_event(:component, id: "1", __version: 1)
 
diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb
index ad5e941f4..4834bde20 100644
--- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb
+++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb
@@ -96,7 +96,31 @@ class Indexer
           end
         end
 
+        it "correlates datastore failures after an adapter normalizes the event envelope" do
+          event = build_upsert_event(:widget, component_ids: ["c1", "c2"]).except(SCHEMA_VERSION_KEY).merge(JSON_SCHEMA_VERSION_KEY => 1)
+          allow(datastore_router).to receive(:bulk) do |ops, **options|
+            DatastoreIndexingRouter::BulkResult.new({"main" => ops.map { |op| [op, Operation::Result.failure_of(op, "overloaded!")] }})
+          end
+
+          failures = process_returning_failures([event])
+
+          expect(failures).not_to be_empty
+          expect(failures.map(&:event)).to all eq(event.except(JSON_SCHEMA_VERSION_KEY).merge(SCHEMA_VERSION_KEY => 1))
+          expect(failures.map(&:operations)).to all eq(failures.first.operations)
+          expect(failures.first.operations.size).to be > 1
+        end
+
         describe "latency metrics" do
+          it "preserves legacy callers' schema versions in latency logs" do
+            component = upsert_event_with_latency_timestamps(:component, 36, 72).except(SCHEMA_VERSION_KEY).merge(JSON_SCHEMA_VERSION_KEY => 1)
+            process([component])
+
+            expect(logged_jsons_of_type("ElasticGraphIndexingLatencies").first).to include(
+              SCHEMA_VERSION_KEY => 1,
+              JSON_SCHEMA_VERSION_KEY => 1
+            )
+          end
+
           it "extracts latency metrics from events" do
             component = upsert_event_with_latency_timestamps(:component, 36, 72)
             address = upsert_event_with_latency_timestamps(:address, 108, 144)
@@ -135,7 +159,20 @@ class Indexer
 
             expect(logged_jsons_of_type("ElasticGraphIndexingLatencies").first).to include(
               "event_id" => "Component:#{component.fetch("id")}@v#{component.fetch("version")}",
-              "message_id" => "m1"
+              "message_id" => "m1",
+              SCHEMA_VERSION_KEY => component.fetch(SCHEMA_VERSION_KEY),
+              # Deprecated alias, kept for existing dashboards and monitors.
+              JSON_SCHEMA_VERSION_KEY => component.fetch(SCHEMA_VERSION_KEY)
+            )
+          end
+
+          it "logs the selected JSON schema version when the event requests none" do
+            component = upsert_event_with_latency_timestamps(:component, 36, 72).except(SCHEMA_VERSION_KEY)
+            process([component])
+
+            expect(logged_jsons_of_type("ElasticGraphIndexingLatencies").first).to include(
+              SCHEMA_VERSION_KEY => 1,
+              JSON_SCHEMA_VERSION_KEY => 1
             )
           end
 
diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/test_support/converters_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/test_support/converters_spec.rb
index bfed1c8f7..39b1f80f8 100644
--- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/test_support/converters_spec.rb
+++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/test_support/converters_spec.rb
@@ -19,7 +19,7 @@ module TestSupport
               "id" => "1",
               "__version" => 1,
               "__typename" => "Widget",
-              "__json_schema_version" => 1,
+              "__schema_version" => 1,
               "field1" => "value1",
               "field2" => "value2"
             }
@@ -30,7 +30,52 @@ module TestSupport
               "version" => 1,
               "type" => "Widget",
               "record" => {"id" => "1", "field1" => "value1", "field2" => "value2"},
-              JSON_SCHEMA_VERSION_KEY => 1
+              SCHEMA_VERSION_KEY => 1
+            )
+          end
+
+          it "accepts the legacy `__json_schema_version` attribute, so factories from older projects keep working" do
+            factory_record = {
+              "id" => "1",
+              "__version" => 1,
+              "__typename" => "Widget",
+              "__json_schema_version" => 3,
+              "field1" => "value1"
+            }
+
+            expect(TestSupport::Converters.upsert_event_for(factory_record)).to eq(
+              "op" => "upsert",
+              "id" => "1",
+              "version" => 1,
+              "type" => "Widget",
+              "record" => {"id" => "1", "field1" => "value1"},
+              SCHEMA_VERSION_KEY => 3
+            )
+          end
+
+          it "preserves invalid versions for validation and gives the generic key precedence" do
+            record = {"id" => "1", "__version" => 1, "__typename" => "Widget", "__schema_version" => false, "__json_schema_version" => 3}
+
+            event = TestSupport::Converters.upsert_event_for(record)
+
+            expect(event.fetch(SCHEMA_VERSION_KEY)).to be false
+            expect(event.fetch("record")).to eq("id" => "1")
+          end
+
+          it "omits the schema version when the factory record supplies none, since the key is optional" do
+            factory_record = {
+              "id" => "1",
+              "__version" => 1,
+              "__typename" => "Widget",
+              "field1" => "value1"
+            }
+
+            expect(TestSupport::Converters.upsert_event_for(factory_record)).to eq(
+              "op" => "upsert",
+              "id" => "1",
+              "version" => 1,
+              "type" => "Widget",
+              "record" => {"id" => "1", "field1" => "value1"}
             )
           end
         end
@@ -41,7 +86,7 @@ module TestSupport
               "id" => "1",
               "__typename" => "Widget",
               "__version" => 1,
-              "__json_schema_version" => 1,
+              "__schema_version" => 1,
               "field1" => "value1",
               "field2" => "value2"
             }
@@ -50,7 +95,7 @@ module TestSupport
               "id" => "2",
               "__typename" => "Address",
               "__version" => 5,
-              "__json_schema_version" => 1,
+              "__schema_version" => 1,
               "field3" => "value5"
             }
 
@@ -63,7 +108,7 @@ module TestSupport
                 "version" => 1,
                 "type" => "Widget",
                 "record" => {"id" => "1", "field1" => "value1", "field2" => "value2"},
-                JSON_SCHEMA_VERSION_KEY => 1
+                SCHEMA_VERSION_KEY => 1
               },
               {
                 "op" => "upsert",
@@ -71,7 +116,7 @@ module TestSupport
                 "version" => 5,
                 "type" => "Address",
                 "record" => {"id" => "2", "field3" => "value5"},
-                JSON_SCHEMA_VERSION_KEY => 1
+                SCHEMA_VERSION_KEY => 1
               }
             ])
           end
diff --git a/elasticgraph-indexer_lambda/spec/unit/elastic_graph/indexer_lambda/sqs_processor_spec.rb b/elasticgraph-indexer_lambda/spec/unit/elastic_graph/indexer_lambda/sqs_processor_spec.rb
index 7d341ff91..cd6e1f19d 100644
--- a/elasticgraph-indexer_lambda/spec/unit/elastic_graph/indexer_lambda/sqs_processor_spec.rb
+++ b/elasticgraph-indexer_lambda/spec/unit/elastic_graph/indexer_lambda/sqs_processor_spec.rb
@@ -6,6 +6,7 @@
 #
 # frozen_string_literal: true
 
+require "aws-sdk-s3"
 require "elastic_graph/errors"
 require "elastic_graph/indexer/failed_event_error"
 require "elastic_graph/indexer/indexing_event_decoder"
@@ -14,7 +15,6 @@
 require "elastic_graph/json_ingestion/indexing_event_decoder"
 require "elastic_graph/spec_support/lambda_function"
 require "json"
-require "aws-sdk-s3"
 
 module ElasticGraph
   module IndexerLambda
@@ -36,7 +36,7 @@ module IndexerLambda
           sqs_processor.process(lambda_event)
 
           expect(indexer_processor).to have_received(:process_returning_failures).with([
-            {"field1" => {}, "message_id" => "a"}
+            {"field1" => {}, "message_id" => "a", INGESTION_FORMAT_KEY => "json"}
           ], refresh_indices: false)
         end
 
@@ -52,9 +52,9 @@ module IndexerLambda
           sqs_processor.process(lambda_event)
 
           expect(indexer_processor).to have_received(:process_returning_failures).with([
-            {"field1" => {}, "message_id" => "a"},
-            {"field2" => {}, "message_id" => "b"},
-            {"field3" => {}, "message_id" => "c"}
+            {"field1" => {}, "message_id" => "a", INGESTION_FORMAT_KEY => "json"},
+            {"field2" => {}, "message_id" => "b", INGESTION_FORMAT_KEY => "json"},
+            {"field3" => {}, "message_id" => "c", INGESTION_FORMAT_KEY => "json"}
           ], refresh_indices: false)
         end
 
@@ -69,11 +69,11 @@ module IndexerLambda
           sqs_processor.process(lambda_event)
 
           expect(indexer_processor).to have_received(:process_returning_failures).with([
-            {"field1" => {}, "message_id" => "a"},
-            {"field2" => {}, "message_id" => "a"},
-            {"field3" => {}, "message_id" => "b"},
-            {"field4" => {}, "message_id" => "b"},
-            {"field5" => {}, "message_id" => "b"}
+            {"field1" => {}, "message_id" => "a", INGESTION_FORMAT_KEY => "json"},
+            {"field2" => {}, "message_id" => "a", INGESTION_FORMAT_KEY => "json"},
+            {"field3" => {}, "message_id" => "b", INGESTION_FORMAT_KEY => "json"},
+            {"field4" => {}, "message_id" => "b", INGESTION_FORMAT_KEY => "json"},
+            {"field5" => {}, "message_id" => "b", INGESTION_FORMAT_KEY => "json"}
           ], refresh_indices: false)
         end
 
@@ -165,7 +165,7 @@ module IndexerLambda
           sqs_processor.process(lambda_event)
 
           expect(indexer_processor).to have_received(:process_returning_failures).with(
-            [event_payload.merge("message_id" => "a")],
+            [event_payload.merge("message_id" => "a", INGESTION_FORMAT_KEY => "json")],
             refresh_indices: false
           )
         end
diff --git a/elasticgraph-json_ingestion/README.md b/elasticgraph-json_ingestion/README.md
index ddd12e2fc..d60b96d91 100644
--- a/elasticgraph-json_ingestion/README.md
+++ b/elasticgraph-json_ingestion/README.md
@@ -106,10 +106,35 @@ end
 
 Beyond schema definition, this gem teaches `elasticgraph-indexer` how to ingest JSON events: it provides an
 ingestion adapter that validates each event against the JSON schema identified by the event's
-`json_schema_version` and prepares its record for indexing using that version's view of the schema.
+`schema_version` and prepares its record for indexing using that version's view of the schema.
 
-No configuration is needed: defining your schema with this gem's `SchemaDefinition::APIExtension` registers
-an indexer extension in your schema artifacts' runtime metadata, which the indexer applies when it boots.
+Defining your schema with this gem's `SchemaDefinition::APIExtension` registers an indexer extension in
+your generated runtime metadata. Install this gem in the indexer deployment and regenerate the artifacts
+to make the adapter available. Encoded transports also need the decoder configuration below.
+
+### Schema versions
+
+The adapter resolves the version of each event as follows:
+
+- The `schema_version` key selects the JSON schema version. When the exact version is unavailable, the
+  adapter selects the closest available version and logs `ElasticGraphMissingJSONSchemaVersion`.
+- The legacy `json_schema_version` key still works, so a publisher or an in-process caller that predates
+  the ingestion-format-neutral key needs no change. If both keys are present, `schema_version` wins
+  for both direct events and decoded payloads. A non-integer value such as `false` is rejected; a missing
+  or null version selects the latest schema.
+- An event that carries neither key gets the latest available JSON schema version. The adapter still
+  validates the event against that version, so a malformed event still fails.
+
+After validation, the adapter returns a normalized copy of the event with `schema_version` set to the
+version it selected and the legacy alias removed. Logs and warehouse partitions use that selected
+version, including when an event omitted its version or requested an unavailable one. The original
+event is not mutated; fallback logs retain both the requested and selected versions.
+
+Publishers should continue sending an explicit JSON schema version and deployments should retain the
+historical artifacts needed to process queued events. Omitting a version ties interpretation to the
+latest schema installed on each indexer. Replaying the same unversioned event after a rename, deletion,
+or validation change can produce a different result or fail validation. Optional versions are useful
+for prototyping; they do not provide the same schema-evolution guarantees as versioned events.
 
 This gem also provides the `be_a_valid_elastic_graph_event` RSpec matcher (via
 `require "elastic_graph/json_ingestion/spec_support/event_matcher"`) for testing that publisher events
@@ -128,8 +153,36 @@ indexer:
     require_path: elastic_graph/json_ingestion/indexing_event_decoder
 ```
 
+The decoder tags events with `ingestion_format: "json"` and leaves version resolution to the adapter.
+This also routes versionless JSON correctly when several adapters are installed. Untagged direct events
+continue to default to JSON; decoders and direct callers for other formats must set their format tag.
+An explicitly different format is never routed to JSON just because it is the only installed adapter.
+
 See the `elasticgraph-indexer` README for the decoder extension interface.
 
+## Upgrading an existing JSON deployment
+
+1. Include `elasticgraph-json_ingestion` in the indexer's runtime bundle, including both indexing and
+   warehouse Lambda packages. A dependency restricted to development or schema generation is insufficient.
+2. Enable `ElasticGraph::JSONIngestion::SchemaDefinition::APIExtension` in your schema definition if it
+   is not already enabled. Run `bundle exec rake schema_artifacts:dump` and deploy the regenerated
+   runtime metadata together with the matching gems. Older runtime metadata does not register the adapter.
+3. Add the `indexer.indexing_event_decoder` configuration shown above to each environment that consumes
+   encoded payloads, including both SQS Lambdas. Direct calls to `indexer.processor.process` do not need a decoder.
+4. Change matcher requires from `elastic_graph/indexer/spec_support/event_matcher` to
+   `elastic_graph/json_ingestion/spec_support/event_matcher`. Custom code that used
+   `indexer.record_preparer_factory` can construct `ElasticGraph::JSONIngestion::RecordPreparerFactory`
+   with `indexer.schema_artifacts` instead.
+5. Keep publishing `json_schema_version`; existing publisher envelopes and versioned JSON schema
+   artifacts remain supported. Latency and warehouse logs include the deprecated `json_schema_version`
+   alias alongside `schema_version`. Both now report the selected version. Warehouse JSON events that
+   omit a version use the selected version's partition; `unversioned` is reserved for genuinely
+   unversioned ingestion formats.
+
+Generated project templates include the decoder setting, but existing project settings are not
+rewritten automatically. Validate an existing publisher event through your deployed transport after
+upgrading, and check its schema version in the indexing or warehouse logs.
+
 ## Dependency Diagram
 
 ```mermaid
diff --git a/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexing_event_decoder.rb b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexing_event_decoder.rb
index 40dac5253..38989e633 100644
--- a/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexing_event_decoder.rb
+++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexing_event_decoder.rb
@@ -6,6 +6,7 @@
 #
 # frozen_string_literal: true
 
+require "elastic_graph/constants"
 require "json"
 
 module ElasticGraph
@@ -13,6 +14,10 @@ module JSONIngestion
     # An indexing event decoder for payloads encoded as newline-delimited JSON objects
     # ([JSON Lines](https://jsonlines.org/)). Configure it via the `indexer.indexing_event_decoder`
     # setting of `elasticgraph-indexer`.
+    #
+    # Tags decoded events with `ingestion_format: "json"` so that the indexer can route them
+    # independently of their schema version. The JSON ingestion adapter resolves version aliases
+    # and selects the schema, consistently for decoded payloads and direct in-process events.
     class IndexingEventDecoder
       # @param config [Hash<String, Object>] configuration from the `indexing_event_decoder.config` setting
       # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts
@@ -24,7 +29,9 @@ def initialize(config:, schema_artifacts:, logger:)
       # @param payload [String] a raw payload from the transport
       # @return [Array<Hash<String, Object>>] the decoded ElasticGraph indexing events
       def decode(payload)
-        payload.split("\n").map { |event| JSON.parse(event) }
+        payload.split("\n").map do |event_json|
+          JSON.parse(event_json).merge(INGESTION_FORMAT_KEY => "json")
+        end
       end
     end
   end
diff --git a/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb
index 7115f786f..68d3c7cc2 100644
--- a/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb
+++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb
@@ -14,10 +14,15 @@
 
 module ElasticGraph
   module JSONIngestion
-    # Ingestion adapter for events in ElasticGraph's versioned JSON format: it validates events
-    # against the JSON schema identified by the event's `json_schema_version`, and prepares
-    # records using that version's view of the schema. Made available to the indexer by the
-    # {IndexerExtension} that {SchemaDefinition::APIExtension} registers.
+    # Ingestion adapter for events in ElasticGraph's JSON format: it validates events against the
+    # JSON schema identified by the event's `schema_version`, and prepares records using that
+    # version's view of the schema. Made available to the indexer by the {IndexerExtension} that
+    # {SchemaDefinition::APIExtension} registers.
+    #
+    # The schema version is optional. An event that omits it gets the latest available JSON schema
+    # version. An event may also carry the legacy `json_schema_version` key instead of
+    # `schema_version`, so that publishers and in-process callers that predate the
+    # ingestion-format-neutral key keep working.
     class IngestionAdapter
       # Shorthand for the result type defined by the indexer's ingestion adapter interface.
       ValidationResult = Indexer::IngestionAdapter::ValidationResult
@@ -33,93 +38,128 @@ def initialize(schema_artifacts:, logger:, configure_record_validator: nil)
         @record_preparer_factory = RecordPreparerFactory.new(schema_artifacts)
       end
 
-      # Indicates whether this adapter recognizes the given event as one of its own, based on the
-      # presence of the `json_schema_version` field in the event envelope.
+      # Recognizes events tagged as JSON by the decoder. Untagged events default to JSON for
+      # compatibility with existing in-process callers, including those that omit a schema version.
+      # Other formats must identify themselves with `ingestion_format`.
       #
       # @param event [Hash<String, Object>] an ElasticGraph indexing event
       # @return [Boolean] whether this adapter handles the event
       def handles_event?(event)
-        event.key?(JSON_SCHEMA_VERSION_KEY)
+        event.fetch(INGESTION_FORMAT_KEY, "json") == "json"
       end
 
       # Validates the given event and resolves the record preparer appropriate for the event's
-      # JSON schema version.
+      # schema version.
       #
       # @param event [Hash<String, Object>] an ElasticGraph indexing event
       # @param skip_record_validation [Boolean] whether to skip record validation; the event envelope must still be validated
       # @return [Indexer::IngestionAdapter::ValidationResult] the result of validating the event
       def validate_event(event, skip_record_validation: false)
-        selected_json_schema_version = select_json_schema_version(event) { |failure| return failure }
+        selected_schema_version = select_schema_version(event) { |failure| return failure }
 
-        # Because the `select_json_schema_version` picks the closest-matching json schema version, the incoming
-        # event might not match the expected json_schema_version value in the json schema (which is a `const` field).
-        # This is by design, since we're picking a schema based on best-effort, so to avoid that by-design validation error,
-        # performing the envelope validation on a "patched" version of the event.
-        event_with_patched_envelope = event.merge({JSON_SCHEMA_VERSION_KEY => selected_json_schema_version})
+        event_with_patched_envelope = event_for_json_schema_validation(event, selected_schema_version)
 
-        if (error_message = validator(EVENT_ENVELOPE_JSON_SCHEMA_NAME, selected_json_schema_version).validate_with_error_message(event_with_patched_envelope))
+        if (error_message = validator(EVENT_ENVELOPE_JSON_SCHEMA_NAME, selected_schema_version).validate_with_error_message(event_with_patched_envelope))
           return ValidationResult.invalid(payload_description: "event payload", message: error_message)
         end
 
         record = event.fetch("record")
         graphql_type_name = event.fetch("type")
 
-        if !skip_record_validation && (error_message = validator(graphql_type_name, selected_json_schema_version).validate_with_error_message(record))
+        if !skip_record_validation && (error_message = validator(graphql_type_name, selected_schema_version).validate_with_error_message(record))
           return ValidationResult.invalid(payload_description: "#{graphql_type_name} record", message: error_message)
         end
 
-        ValidationResult.valid(@record_preparer_factory.for_json_schema_version(selected_json_schema_version))
+        ValidationResult.valid(
+          @record_preparer_factory.for_json_schema_version(selected_schema_version),
+          event: event.except(JSON_SCHEMA_VERSION_KEY).merge(SCHEMA_VERSION_KEY => selected_schema_version)
+        )
       end
 
       private
 
-      def select_json_schema_version(event)
-        available_json_schema_versions = @schema_artifacts.available_json_schema_versions
+      # The JSON schemas expect the `json_schema_version` key, but the event carries the
+      # ingestion-format-neutral `schema_version` key, so restore the JSON-specific key here. The
+      # value is the selected version rather than the requested one, because the envelope schema
+      # declares `json_schema_version` as a `const`, and `select_schema_version` selects the
+      # closest available version by design. This also supplies the key for an event that omitted
+      # its version, since the envelope schema requires the key.
+      def event_for_json_schema_validation(event, selected_schema_version)
+        event
+          .except(INGESTION_FORMAT_KEY, SCHEMA_VERSION_KEY, JSON_SCHEMA_VERSION_KEY)
+          .merge(JSON_SCHEMA_VERSION_KEY => selected_schema_version)
+      end
 
-        requested_json_schema_version = event[JSON_SCHEMA_VERSION_KEY]
+      # Reads the version the event requests. The ingestion-format-neutral `schema_version` key wins,
+      # and the legacy JSON-specific `json_schema_version` key acts as a fallback. A `nil` result
+      # means the event requests no particular version.
+      def requested_schema_version_for(event)
+        event.fetch(SCHEMA_VERSION_KEY) { event[JSON_SCHEMA_VERSION_KEY] }
+      end
 
-        # First check that a valid value has been requested (a positive integer)
-        if !event.key?(JSON_SCHEMA_VERSION_KEY)
-          yield ValidationResult.invalid(payload_description: JSON_SCHEMA_VERSION_KEY, message: "Event lacks a `#{JSON_SCHEMA_VERSION_KEY}`")
-        elsif !requested_json_schema_version.is_a?(Integer) || requested_json_schema_version < 1
-          yield ValidationResult.invalid(payload_description: JSON_SCHEMA_VERSION_KEY, message: "#{JSON_SCHEMA_VERSION_KEY} (#{requested_json_schema_version}) must be a positive integer.")
-        end
+      def select_schema_version(event)
+        available_schema_versions = available_schema_versions_descending
+        requested_schema_version = requested_schema_version_for(event)
+
+        if requested_schema_version.nil?
+          # The schema version is optional, so an event may omit it. Use the latest available version,
+          # which is the first entry of the descending list. The event still gets validated against
+          # that version's JSON schemas, so a malformed event still fails.
+          selected_schema_version = available_schema_versions.first
+        else
+          # Check that a valid value has been requested (a positive integer).
+          unless requested_schema_version.is_a?(Integer) && requested_schema_version >= 1
+            yield ValidationResult.invalid(
+              payload_description: SCHEMA_VERSION_KEY,
+              message: "#{SCHEMA_VERSION_KEY} (#{requested_schema_version}) must be a positive integer."
+            )
+          end
 
-        # The requested version might not necessarily be available (if the publisher is deployed ahead of the indexer, or an old schema
-        # version is removed prematurely, or an indexer deployment is rolled back). So the behavior is to always pick the closest-available
-        # version. If there's an exact match, great. Even if not an exact match, if the incoming event payload conforms to the closest match,
-        # the event can still be indexed.
-        #
-        # This min_by block will take the closest version in the list. If a tie occurs, the first value in the list wins. The desired
-        # behavior is in the event of a tie (highly unlikely, there shouldn't be a gap in available json schema versions), the higher version
-        # should be selected. So to get that behavior, the list is sorted in descending order.
-        #
-        selected_json_schema_version = available_json_schema_versions.sort.reverse.min_by { |version| (requested_json_schema_version - version).abs }
-
-        if selected_json_schema_version != requested_json_schema_version
-          @logger.info({
-            "message_type" => "ElasticGraphMissingJSONSchemaVersion",
-            "message_id" => event["message_id"],
-            "event_id" => Indexer::EventID.from_event(event),
-            "event_type" => event["type"],
-            "requested_json_schema_version" => requested_json_schema_version,
-            "selected_json_schema_version" => selected_json_schema_version
-          })
+          # The requested version might not necessarily be available (if the publisher is deployed ahead of the indexer, or an old schema
+          # version is removed prematurely, or an indexer deployment is rolled back). So the behavior is to always pick the closest-available
+          # version. If there's an exact match, great. Even if not an exact match, if the incoming event payload conforms to the closest match,
+          # the event can still be indexed.
+          #
+          # This min_by block will take the closest version in the list. If a tie occurs, the first value in the list wins. The desired
+          # behavior is in the event of a tie (highly unlikely, there shouldn't be a gap in available json schema versions), the higher version
+          # should be selected. So to get that behavior, the list is sorted in descending order.
+          #
+          selected_schema_version = available_schema_versions.min_by { |version| (requested_schema_version - version).abs }
+
+          if selected_schema_version != requested_schema_version
+            @logger.info({
+              "message_type" => "ElasticGraphMissingJSONSchemaVersion",
+              "message_id" => event["message_id"],
+              "event_id" => Indexer::EventID.from_event(event),
+              "event_type" => event["type"],
+              # These fields keep their JSON-specific names to match the JSON-specific message type,
+              # so that dashboards and monitors that watch them keep working.
+              "requested_json_schema_version" => requested_schema_version,
+              "selected_json_schema_version" => selected_schema_version
+            })
+          end
         end
 
-        if selected_json_schema_version.nil?
+        if selected_schema_version.nil?
           yield ValidationResult.invalid(
-            payload_description: JSON_SCHEMA_VERSION_KEY,
-            message: "Failed to select json schema version. Requested version: #{event[JSON_SCHEMA_VERSION_KEY]}. \
-            Available json schema versions: #{available_json_schema_versions.sort.join(", ")}"
+            payload_description: SCHEMA_VERSION_KEY,
+            message: "Failed to select schema version. Requested version: #{requested_schema_version.inspect}. \
+            Available schema versions: #{available_schema_versions.sort.join(", ")}"
           )
         end
 
-        selected_json_schema_version
+        selected_schema_version
+      end
+
+      # The available versions, sorted once in descending order. Sorting here rather than per event
+      # avoids repeated work, and the descending order gives `min_by` the tie-break behavior it
+      # needs: on a tie, the higher version wins.
+      def available_schema_versions_descending
+        @available_schema_versions_descending ||= @schema_artifacts.available_json_schema_versions.sort.reverse
       end
 
-      def validator(type, selected_json_schema_version)
-        factory = validator_factories_by_version[selected_json_schema_version] # : Support::JSONSchema::ValidatorFactory
+      def validator(type, selected_schema_version)
+        factory = validator_factories_by_version[selected_schema_version] # : Support::JSONSchema::ValidatorFactory
         factory.validator_for(type)
       end
 
diff --git a/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/ingestion_adapter.rbs b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/ingestion_adapter.rbs
index 20d62d35d..5865c58c3 100644
--- a/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/ingestion_adapter.rbs
+++ b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/ingestion_adapter.rbs
@@ -19,9 +19,14 @@ module ElasticGraph
 
       private
 
-      def select_json_schema_version: (Indexer::event) { (Indexer::IngestionAdapter::ValidationResult) -> bot } -> (::Integer | bot)
+      def event_for_json_schema_validation: (Indexer::event, ::Integer) -> ::Hash[::String, untyped]
+      def requested_schema_version_for: (Indexer::event) -> untyped
+      def select_schema_version: (Indexer::event) { (Indexer::IngestionAdapter::ValidationResult) -> bot } -> (::Integer | bot)
       def validator: (::String, ::Integer) -> Support::JSONSchema::Validator
 
+      @available_schema_versions_descending: ::Array[::Integer]?
+      def available_schema_versions_descending: () -> ::Array[::Integer]
+
       @validator_factories_by_version: ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]?
       def validator_factories_by_version: () -> ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]
 
diff --git a/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexing_event_decoder_spec.rb b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexing_event_decoder_spec.rb
index 93c74cba6..b5c3c123b 100644
--- a/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexing_event_decoder_spec.rb
+++ b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexing_event_decoder_spec.rb
@@ -21,8 +21,16 @@ module JSONIngestion
         JSONL
 
         expect(decoder.decode(payload)).to eq([
-          {"op" => "upsert", "id" => "1"},
-          {"op" => "upsert", "id" => "2"}
+          {"op" => "upsert", "id" => "1", INGESTION_FORMAT_KEY => "json"},
+          {"op" => "upsert", "id" => "2", INGESTION_FORMAT_KEY => "json"}
+        ])
+      end
+
+      it "preserves version keys for the ingestion adapter to resolve" do
+        decoder = IndexingEventDecoder.new(config: {}, schema_artifacts: nil, logger: nil) # args are not used
+
+        expect(decoder.decode('{"op":"upsert","id":"1","json_schema_version":3,"schema_version":2}')).to eq([
+          {"op" => "upsert", "id" => "1", JSON_SCHEMA_VERSION_KEY => 3, SCHEMA_VERSION_KEY => 2, INGESTION_FORMAT_KEY => "json"}
         ])
       end
 
diff --git a/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb
index d7d3e159a..8eaeb56fa 100644
--- a/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb
+++ b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb
@@ -7,6 +7,7 @@
 # frozen_string_literal: true
 
 require "elastic_graph/constants"
+require "elastic_graph/json_ingestion/indexing_event_decoder"
 require "elastic_graph/json_ingestion/ingestion_adapter"
 
 module ElasticGraph
@@ -16,21 +17,62 @@ module JSONIngestion
       let(:adapter) { build_adapter }
 
       describe "#handles_event?" do
-        it "recognizes events that have a `#{JSON_SCHEMA_VERSION_KEY}` in their envelope" do
+        it "recognizes events that have a `#{SCHEMA_VERSION_KEY}` in their envelope" do
           event = build_upsert_event(:component)
 
           expect(adapter.handles_event?(event)).to be true
-          expect(adapter.handles_event?(event.except(JSON_SCHEMA_VERSION_KEY))).to be false
         end
+
+        it "recognizes events that have a legacy `#{JSON_SCHEMA_VERSION_KEY}` in their envelope" do
+          event = build_upsert_event(:component)
+          legacy_event = event.except(SCHEMA_VERSION_KEY).merge(JSON_SCHEMA_VERSION_KEY => event.fetch(SCHEMA_VERSION_KEY))
+
+          expect(adapter.handles_event?(legacy_event)).to be true
+        end
+
+        it "recognizes untagged JSON events even without a schema version" do
+          event = build_upsert_event(:component)
+
+          expect(adapter.handles_event?(event.except(SCHEMA_VERSION_KEY))).to be true
+        end
+      end
+
+      it "recognizes JSON tags and rejects other formats regardless of their schema version" do
+        event = build_upsert_event(:component)
+
+        expect(adapter.handles_event?(event.merge(INGESTION_FORMAT_KEY => "json"))).to be true
+        expect(adapter.handles_event?(event.merge(INGESTION_FORMAT_KEY => "other"))).to be false
+        expect(adapter.handles_event?(event.merge(INGESTION_FORMAT_KEY => nil))).to be false
       end
 
       describe "#validate_event" do
+        it "returns a normalized legacy envelope without mutating the caller's event" do
+          event = build_upsert_event(:component).except(SCHEMA_VERSION_KEY).merge(
+            JSON_SCHEMA_VERSION_KEY => 1, "message_id" => "m1", INGESTION_FORMAT_KEY => "json"
+          ).freeze
+
+          result = adapter.validate_event(event)
+
+          expect(result.failure).to be nil
+          expect(result.event).to eq(event.except(JSON_SCHEMA_VERSION_KEY).merge(SCHEMA_VERSION_KEY => 1))
+          expect(event).not_to have_key(SCHEMA_VERSION_KEY)
+        end
+
+        [SCHEMA_VERSION_KEY, JSON_SCHEMA_VERSION_KEY].each do |key|
+          it "rejects a false #{key} instead of interpreting it as an absent version" do
+            event = build_upsert_event(:component).except(SCHEMA_VERSION_KEY).merge(key => false)
+
+            expect_invalid(event, payload_description: SCHEMA_VERSION_KEY, message_including: ["must be a positive integer", "false"])
+          end
+        end
+
         it "returns a valid result with a record preparer for the event's JSON schema version" do
           event = build_upsert_event(:component)
 
           result = adapter.validate_event(event)
 
           expect(result.failure).to be nil
+          expect(result.event.fetch(SCHEMA_VERSION_KEY)).to eq(1)
           expect(result.record_preparer.prepare_for_index("Component", {"id" => "1", "unknown_field" => 3}, {}))
             .to eq({"id" => "1"})
         end
@@ -100,10 +142,40 @@ module JSONIngestion
           expect_invalid(event, payload_description: "event payload", message_including: ["missing_keys", "version"])
         end
 
-        it "notifies an error on missing `#{JSON_SCHEMA_VERSION_KEY}`" do
-          event = build_upsert_event(:component).except(JSON_SCHEMA_VERSION_KEY)
+        it "accepts an event that carries no `#{SCHEMA_VERSION_KEY}`" do
+          event = build_upsert_event(:component).except(SCHEMA_VERSION_KEY)
+
+          result = adapter.validate_event(event)
+
+          expect(result.failure).to be nil
+          expect(result.event.fetch(SCHEMA_VERSION_KEY)).to eq(1)
+          expect(result.record_preparer.prepare_for_index("Component", {"id" => "1", "unknown_field" => 3}, {}))
+            .to eq({"id" => "1"})
+        end
+
+        it "still validates the record of an event that carries no `#{SCHEMA_VERSION_KEY}`" do
+          event = build_upsert_event(:component, id: "1", __version: 1).except(SCHEMA_VERSION_KEY)
+          event["record"]["name"] = 123
+
+          expect_invalid(event, payload_description: "Component record", message_including: ["name"])
+        end
+
+        it "accepts the legacy `#{JSON_SCHEMA_VERSION_KEY}` key in place of `#{SCHEMA_VERSION_KEY}`" do
+          event = build_upsert_event(:component)
+          legacy_event = event.except(SCHEMA_VERSION_KEY).merge(JSON_SCHEMA_VERSION_KEY => event.fetch(SCHEMA_VERSION_KEY))
+
+          result = adapter.validate_event(legacy_event)
 
-          expect_invalid(event, payload_description: JSON_SCHEMA_VERSION_KEY, message_including: ["Event lacks a `#{JSON_SCHEMA_VERSION_KEY}`"])
+          expect(result.failure).to be nil
+          expect(result.event.fetch(SCHEMA_VERSION_KEY)).to eq(1)
+          expect(result.record_preparer.prepare_for_index("Component", {"id" => "1", "unknown_field" => 3}, {}))
+            .to eq({"id" => "1"})
+        end
+
+        it "notifies an error on a `#{SCHEMA_VERSION_KEY}` that is not a positive integer" do
+          event = build_upsert_event(:component).merge(SCHEMA_VERSION_KEY => "not a version")
+
+          expect_invalid(event, payload_description: SCHEMA_VERSION_KEY, message_including: ["must be a positive integer", "not a version"])
         end
 
         it "notifies an error when given a record that does not satisfy the type's JSON schema, while avoiding revealing PII" do
@@ -135,6 +207,7 @@ module JSONIngestion
               2 => schema_artifacts.json_schemas_for(1),
               4 => ::Marshal.load(::Marshal.dump(schema_artifacts.json_schemas_for(1))).tap do |schema|
                 schema["$defs"]["Color"]["enum"] << "YELLOW"
+                schema["$defs"]["Widget"]["properties"]["name"]["ElasticGraph"]["nameInIndex"] = "renamed_name"
               end
             }
 
@@ -148,16 +221,44 @@ module JSONIngestion
             end
           end
 
+          it "uses the same version precedence for direct events and decoded payloads" do
+            decoder = IndexingEventDecoder.new(config: {}, schema_artifacts: schema_artifacts, logger: logger)
+            event = build_upsert_event(:widget, __schema_version: 2).merge(JSON_SCHEMA_VERSION_KEY => 4)
+            event["record"]["options"]["color"] = "YELLOW"
+
+            direct_result = adapter.validate_event(event)
+            decoded_result = adapter.validate_event(decoder.decode(::JSON.generate(event)).first)
+
+            expect(direct_result.failure.message).to include("/options/color")
+            expect(decoded_result.failure).to eq(direct_result.failure)
+          end
+
+          it "does not replace an invalid generic version with a valid legacy version" do
+            event = build_upsert_event(:widget, __schema_version: false).merge(JSON_SCHEMA_VERSION_KEY => 4)
+
+            expect_invalid(event, payload_description: SCHEMA_VERSION_KEY, message_including: ["must be a positive integer", "false"])
+          end
+
+          it "resolves an absent version to the latest schema and exposes it downstream" do
+            event = build_upsert_event(:widget).except(SCHEMA_VERSION_KEY)
+            event["record"]["options"]["color"] = "YELLOW"
+
+            result = adapter.validate_event(event)
+
+            expect(result.failure).to be nil
+            expect(result.event.fetch(SCHEMA_VERSION_KEY)).to eq(4)
+          end
+
           it "validates against an older version of a json schema if specified" do
             # YELLOW doesn't exist in schema version 2. So expect an error when json_schema_version is set to 2.
-            event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 2)
+            event = build_upsert_event(:widget, id: "1", __version: 1, __schema_version: 2)
             event["record"]["options"]["color"] = "YELLOW"
 
             expect_invalid(event, payload_description: "Widget record", message_including: ["/options/color"])
           end
 
           it "validates against the latest version of a json schema if specified" do
-            event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 4)
+            event = build_upsert_event(:widget, id: "1", __version: 1, __schema_version: 4)
             event["record"]["options"]["color"] = "YELLOW"
 
             expect(adapter.validate_event(event).failure).to be nil
@@ -165,10 +266,12 @@ module JSONIngestion
 
           it "validates against the closest version if the requested version is newer than what's available" do
             # 5 is closest to "4", validation should match behavior from version "4" - YELLOW should pass validation.
-            event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 5)
+            event = build_upsert_event(:widget, id: "1", __version: 1, __schema_version: 5)
             event["record"]["options"]["color"] = "YELLOW"
 
-            expect(adapter.validate_event(event).failure).to be nil
+            result = adapter.validate_event(event)
+            expect(result.failure).to be nil
+            expect(result.event.fetch(SCHEMA_VERSION_KEY)).to eq(4)
 
             expect(logged_jsons_of_type("ElasticGraphMissingJSONSchemaVersion").last).to include(
               "event_id" => "Widget:1@v1",
@@ -178,9 +281,20 @@ module JSONIngestion
             )
           end
 
+          it "prepares a record with the selected fallback schema's field metadata" do
+            event = build_upsert_event(:widget, __schema_version: 1)
+
+            result = adapter.validate_event(event)
+
+            expect(result.failure).to be nil
+            expect(result.event.fetch(SCHEMA_VERSION_KEY)).to eq(2)
+            expect(result.record_preparer.prepare_for_index("Widget", {"id" => "1", "name" => "old name"}, {}))
+              .to eq({"id" => "1", "name" => "old name"})
+          end
+
           it "validates against the closest version if the requested version is older than what's available" do
             # 1 is closest to "2", validation should match behavior from version "2" - YELLOW should fail validation.
-            event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 1).merge("message_id" => "m123")
+            event = build_upsert_event(:widget, id: "1", __version: 1, __schema_version: 1).merge("message_id" => "m123")
             event["record"]["options"]["color"] = "YELLOW"
 
             # Should fail, but should still log the version mismatch as well.
@@ -196,10 +310,12 @@ module JSONIngestion
           end
 
           it "validates against a version newer than what's requested, if the requested version is equidistant from two available versions" do
-            event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 3)
+            event = build_upsert_event(:widget, id: "1", __version: 1, __schema_version: 3)
             event["record"]["options"]["color"] = "YELLOW"
 
-            expect(adapter.validate_event(event).failure).to be nil
+            result = adapter.validate_event(event)
+            expect(result.failure).to be nil
+            expect(result.event.fetch(SCHEMA_VERSION_KEY)).to eq(4)
 
             expect(logged_jsons_of_type("ElasticGraphMissingJSONSchemaVersion").last).to include(
               "event_id" => "Widget:1@v1",
@@ -210,9 +326,35 @@ module JSONIngestion
           end
 
           it "notifies an error if an invalid (e.g. negative) json_schema_version is specified" do
-            event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: -1)
+            event = build_upsert_event(:widget, id: "1", __version: 1, __schema_version: -1)
 
-            expect_invalid(event, payload_description: JSON_SCHEMA_VERSION_KEY, message_including: ["must be a positive integer", "(-1)"])
+            expect_invalid(event, payload_description: SCHEMA_VERSION_KEY, message_including: ["must be a positive integer", "(-1)"])
+          end
+
+          it "uses the latest available version when the event carries no schema version" do
+            # YELLOW exists only in version 4, so it passes validation only if version 4 was selected.
+            event = build_upsert_event(:widget, id: "1", __version: 1).except(SCHEMA_VERSION_KEY)
+            event["record"]["options"]["color"] = "YELLOW"
+
+            expect(adapter.validate_event(event).failure).to be nil
+          end
+
+          it "does not log a version mismatch when the event carries no schema version" do
+            event = build_upsert_event(:widget, id: "1", __version: 1).except(SCHEMA_VERSION_KEY)
+
+            adapter.validate_event(event)
+
+            expect(logged_jsons_of_type("ElasticGraphMissingJSONSchemaVersion")).to be_empty
+          end
+
+          it "honors the legacy `#{JSON_SCHEMA_VERSION_KEY}` key when selecting the version" do
+            # Requesting version 2 via the legacy key must reject YELLOW, which only version 4 allows.
+            event = build_upsert_event(:widget, id: "1", __version: 1)
+              .except(SCHEMA_VERSION_KEY)
+              .merge(JSON_SCHEMA_VERSION_KEY => 2)
+            event["record"]["options"]["color"] = "YELLOW"
+
+            expect_invalid(event, payload_description: "Widget record", message_including: ["/options/color"])
           end
         end
 
@@ -221,7 +363,15 @@ module JSONIngestion
 
           event = build_upsert_event(:component, id: "1", __version: 1)
 
-          expect_invalid(event, payload_description: JSON_SCHEMA_VERSION_KEY, message_including: ["Failed to select json schema version"])
+          expect_invalid(event, payload_description: SCHEMA_VERSION_KEY, message_including: ["Failed to select schema version"])
+        end
+
+        it "notifies an error if no versions are available and the event requests none" do
+          allow(schema_artifacts).to receive(:available_json_schema_versions).and_return(Set[])
+
+          event = build_upsert_event(:component, id: "1", __version: 1).except(SCHEMA_VERSION_KEY)
+
+          expect_invalid(event, payload_description: SCHEMA_VERSION_KEY, message_including: ["Failed to select schema version", "nil"])
         end
 
         def expect_invalid(event, payload_description:, message_including:)
diff --git a/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml b/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml
index d4b36025d..944d7c7f1 100644
--- a/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml
+++ b/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml
@@ -510,7 +510,7 @@ properties:
           - ABC12345678
       skip_record_validation_percents_by_type:
         description: |-
-          Map of GraphQL type names to the percentage of records of that type whose per-record JSON schema validation should be skipped. `0` (or an absent key) validates every record of the type; `100` skips every record; values in between sample, and may be fractional. The decision is deterministic per event id (`type:id@vversion`), so the same event makes the same choice on every retry and on every indexer pod. The event envelope (op, id, type, version, json_schema_version, latency_timestamps) is always validated, regardless of this setting.
+          Map of GraphQL type names to the percentage of records of that type whose per-record schema validation should be skipped. `0` (or an absent key) validates every record of the type; `100` skips every record; values in between sample, and may be fractional. The decision is deterministic per event id (`type:id@vversion`), so the same event makes the same choice on every retry and on every indexer pod. The ingestion adapter always validates the event envelope, including any format-specific schema version, regardless of this setting.
 
           With a large schema the per-record schema walk consumes a significant share of indexing CPU: every record is checked against every regex, enum, min/max, format, and abstract-type discriminator defined for its type. Skipping it trades that check for throughput, which is worthwhile when backfilling data that was already validated upstream. Leaving a percentage of records validated keeps a canary in place so schema drift still surfaces.
 
diff --git a/elasticgraph-support/lib/elastic_graph/constants.rb b/elasticgraph-support/lib/elastic_graph/constants.rb
index b039efd0e..cb76daeec 100644
--- a/elasticgraph-support/lib/elastic_graph/constants.rb
+++ b/elasticgraph-support/lib/elastic_graph/constants.rb
@@ -137,6 +137,14 @@ module ElasticGraph
   # @private
   JSON_SCHEMA_VERSION_KEY = "json_schema_version"
 
+  # Name for field in indexing events that identifies the schema artifact version to use.
+  # @private
+  SCHEMA_VERSION_KEY = "schema_version"
+
+  # The key that identifies the ingestion format of a decoded event.
+  # @return [String]
+  INGESTION_FORMAT_KEY = "ingestion_format"
+
   # String that goes in the middle of a rollover index name, used to mark it as a rollover
   # index (and split on to parse a rollover index name).
   # @private
diff --git a/elasticgraph-support/sig/elastic_graph/constants.rbs b/elasticgraph-support/sig/elastic_graph/constants.rbs
index 7d8e55585..3063b6139 100644
--- a/elasticgraph-support/sig/elastic_graph/constants.rbs
+++ b/elasticgraph-support/sig/elastic_graph/constants.rbs
@@ -22,6 +22,8 @@ module ElasticGraph
   ROLLOVER_INDEX_INFIX_MARKER: ::String
   JSON_SCHEMAS_BY_VERSION_DIRECTORY: ::String
   JSON_SCHEMA_VERSION_KEY: ::String
+  SCHEMA_VERSION_KEY: ::String
+  INGESTION_FORMAT_KEY: ::String
   DERIVED_INDEX_FAILURE_MESSAGE_PREAMBLE: ::String
   INDEX_DATA_UPDATE_SCRIPT_ID: ::String
   UPDATE_WAS_NOOP_MESSAGE_PREAMBLE: ::String
diff --git a/elasticgraph-warehouse_lambda/README.md b/elasticgraph-warehouse_lambda/README.md
index 62dd339a2..0cf492c51 100644
--- a/elasticgraph-warehouse_lambda/README.md
+++ b/elasticgraph-warehouse_lambda/README.md
@@ -4,7 +4,7 @@ Write ElasticGraph-shaped JSONL files to S3, packaged for AWS Lambda.
 
 This gem adapts ElasticGraph's indexing pipeline so that, instead of writing to the datastore,
 it writes batched, gzipped [JSON Lines](https://jsonlines.org/) (JSONL) files to Amazon S3. Each line in the file
-conforms to a specific JSON Schema version for the corresponding object type, with files partitioned by schema version.
+conforms to a specific schema version for the corresponding object type, with files partitioned by schema version.
 
 **Note:** This code does not deduplicate when writing to S3, so the data will contain all events
 and versions published, plus any Lambda retries. Consumers of the S3 bucket are responsible for
@@ -33,10 +33,10 @@ graph LR;
 
 ## What it does
 
-- Consumes ElasticGraph indexing operations and groups them by GraphQL type and JSON schema version
+- Consumes ElasticGraph indexing operations and groups them by GraphQL type and schema version
 - Transforms each operation into a flattened JSON document that matches your ElasticGraph schema
-- Writes one gzipped JSONL file per type per JSON schema version per batch to S3 with deterministic keys:
-  - `s3://<bucket>/<s3_path_prefix>/<TypeName>/v<json_schema_version>/<YYYY-MM-DD>/<uuid>.jsonl.gz`
+- Writes one gzipped JSONL file per type per schema version per batch to S3 with deterministic keys:
+  - `s3://<bucket>/<s3_path_prefix>/<TypeName>/v<schema_version>/<YYYY-MM-DD>/<uuid>.jsonl.gz`
 - Emits structured logs for observability (counts, sizes, S3 key, etc.)
 
 ## When to use it
@@ -67,16 +67,17 @@ warehouse:
 Files are written with the following S3 key format:
 
 ```
-<s3_path_prefix>/<TypeName>/v<json_schema_version>/<YYYY-MM-DD>/<uuid>.jsonl.gz
+<s3_path_prefix>/<TypeName>/v<schema_version>/<YYYY-MM-DD>/<uuid>.jsonl.gz
 ```
 
 - **s3_path_prefix**: Configurable in YAML (warehouse.s3_path_prefix). This is the full prefix you control,
   so you can organize your data however you like (e.g., "dumped-data/Data001" or "prod/analytics/v2").
 - **TypeName**: GraphQL type from the ElasticGraph event
-- **json_schema_version**: The JSON Schema version **selected based on the ingested event's requested version**
+- **schema_version**: The schema version **selected based on the ingested event's requested version**
   (or the closest available version if the exact version isn't available). This ensures data partitioning
   matches the actual schema version used to process each event, making it easier to handle schema evolution
-  and version-specific data processing.
+  and version-specific data processing. An ingestion format with no schema versions gets the fixed segment
+  `unversioned` in place of `v<schema_version>`, so the segment count of the key stays the same.
 - **YYYY-MM-DD**: UTC date when the batch was processed (aligns with common data warehouse
   partitioning strategies)
 - **uuid**: A random UUID for uniqueness
diff --git a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb
index baaa444f5..24ad54a57 100644
--- a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb
+++ b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb
@@ -25,6 +25,10 @@ class WarehouseDumper
       # @return [String] message type for logging when a file is dumped to S3
       LOG_MSG_DUMPED_FILE = "DumpedToWarehouseFile"
 
+      # @return [String] S3 key segment used in place of `v<version>` for an ingestion format that
+      #   has no schema versions
+      UNVERSIONED_S3_KEY_SEGMENT = "unversioned"
+
       def initialize(logger:, s3_client:, s3_bucket_name:, s3_file_prefix:, clock:)
         @logger = logger
         @s3_client = s3_client
@@ -34,20 +38,21 @@ def initialize(logger:, s3_client:, s3_bucket_name:, s3_file_prefix:, clock:)
       end
 
       # Processes a batch of indexing operations by dumping them to S3 as gzipped JSONL files.
-      # Operations are grouped by GraphQL type and JSON schema version, with each group written to a separate file.
+      # Operations are grouped by GraphQL type and schema version, with each group written to a separate file.
       #
       # @param operations [Array<Operation>] the indexing operations to process
       # @param refresh [Boolean] ignored (included for interface compatibility with DatastoreIndexingRouter)
       # @return [BulkResult] result containing success status for all operations
       def bulk(operations, refresh: false)
-        operations_by_type_and_json_schema_version = operations.group_by { |op| [op.event.fetch("type"), op.event.fetch(JSON_SCHEMA_VERSION_KEY)] }
+        # The schema version is optional, since an ingestion format may have no versions at all.
+        operations_by_type_and_schema_version = operations.group_by { |op| [op.event.fetch("type"), op.event[SCHEMA_VERSION_KEY]] }
 
         @logger.info({
           "message_type" => LOG_MSG_RECEIVED_BATCH,
-          "record_counts_by_type" => operations_by_type_and_json_schema_version.transform_keys { |(type, _json_schema_version)| type }.transform_values(&:size)
+          "record_counts_by_type" => operations_by_type_and_schema_version.transform_keys { |(type, _schema_version)| type }.transform_values(&:size)
         })
 
-        operations_by_type_and_json_schema_version.each do |(type, json_schema_version), operations|
+        operations_by_type_and_schema_version.each do |(type, schema_version), operations|
           # Operations coming from the indexer are always Update operations for warehouse dumping
           update_operations = operations # : ::Array[::ElasticGraph::Indexer::Operation::Update]
           jsonl_data = build_jsonl_file_from(update_operations)
@@ -56,7 +61,7 @@ def bulk(operations, refresh: false)
           next if jsonl_data.empty?
 
           gzip_data = compress(jsonl_data)
-          s3_key = generate_s3_key_for(type, json_schema_version)
+          s3_key = generate_s3_key_for(type, schema_version)
 
           # Use if_none_match: "*" to prevent overwrites (defense-in-depth, though UUIDs make collisions impossible)
           @s3_client.put_object(
@@ -72,7 +77,10 @@ def bulk(operations, refresh: false)
             "s3_bucket" => @s3_bucket_name,
             "s3_key" => s3_key,
             "type" => type,
-            JSON_SCHEMA_VERSION_KEY => json_schema_version,
+            SCHEMA_VERSION_KEY => schema_version,
+            # Deprecated alias of `schema_version`, kept so that dashboards and monitors that watch
+            # the old name keep working.
+            JSON_SCHEMA_VERSION_KEY => schema_version,
             "record_count" => operations.size,
             "json_size" => jsonl_data.bytesize,
             "gzip_size" => gzip_data.bytesize
@@ -97,14 +105,18 @@ def source_event_versions_in_index(operations)
 
       private
 
-      def generate_s3_key_for(type, json_schema_version)
+      def generate_s3_key_for(type, schema_version)
         date = @clock.now.utc.strftime("%Y-%m-%d")
         uuid = ::SecureRandom.uuid
 
+        # An ingestion format with no schema versions gets a fixed segment in place of `v<version>`.
+        # The segment count stays the same, so a reader that splits the key keeps working.
+        version_segment = schema_version.nil? ? UNVERSIONED_S3_KEY_SEGMENT : "v#{schema_version}"
+
         [
           @s3_file_prefix,
           type,
-          "v#{json_schema_version}",
+          version_segment,
           date,
           "#{uuid}.jsonl.gz"
         ].join("/")
diff --git a/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs
index 63a71a5de..8f7a4f978 100644
--- a/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs
+++ b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs
@@ -5,6 +5,7 @@ module ElasticGraph
 
       LOG_MSG_RECEIVED_BATCH: ::String
       LOG_MSG_DUMPED_FILE: ::String
+      UNVERSIONED_S3_KEY_SEGMENT: ::String
 
       def initialize: (
         logger: ::Logger,
@@ -22,7 +23,7 @@ module ElasticGraph
       @s3_file_prefix: ::String
       @clock: singleton(::Time)
 
-      def generate_s3_key_for: (::String, ::Integer) -> ::String
+      def generate_s3_key_for: (::String, ::Integer?) -> ::String
 
       def build_jsonl_file_from: (
         ::Array[Indexer::Operation::Update]
diff --git a/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb
index 47109d580..f8ea44377 100644
--- a/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb
+++ b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb
@@ -9,8 +9,8 @@
 require "aws-sdk-s3"
 require "elastic_graph/indexer/operation/update"
 require "elastic_graph/warehouse_lambda/warehouse_dumper"
-require "support/builds_warehouse_lambda"
 require "elastic_graph/spec_support/builds_indexer_operation"
+require "support/builds_warehouse_lambda"
 
 module ElasticGraph
   class WarehouseLambda
@@ -30,22 +30,38 @@ class WarehouseLambda
           "type" => "Widget",
           "id" => "1",
           "version" => 3,
-          "json_schema_version" => 1,
+          SCHEMA_VERSION_KEY => 1,
           "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}
         })
       end
 
+      it "partitions legacy, versionless, and fallback JSON events by the schema actually selected", :factories do
+        legacy = build_upsert_event(:component, id: "legacy").except(SCHEMA_VERSION_KEY).merge(JSON_SCHEMA_VERSION_KEY => 1)
+        versionless = build_upsert_event(:component, id: "versionless").except(SCHEMA_VERSION_KEY)
+        fallback = build_upsert_event(:component, id: "fallback").merge(SCHEMA_VERSION_KEY => 99)
+
+        warehouse_lambda.processor.process([legacy, versionless, fallback])
+
+        uploads = s3_client.api_requests.select { |request| request[:operation_name] == :put_object }
+        expect(uploads.size).to eq(1)
+        expect(uploads.first.fetch(:params).fetch(:key)).to start_with("Data0001/Component/v1/")
+        expect(logged_jsons_of_type(WarehouseDumper::LOG_MSG_DUMPED_FILE)).to match [
+          a_hash_including(SCHEMA_VERSION_KEY => 1, "record_count" => 3)
+        ]
+        expect(logged_jsons_of_type("ElasticGraphIndexingLatencies")).to all include(SCHEMA_VERSION_KEY => 1)
+      end
+
       it "writes operations to S3 as gzipped JSONL files and returns success results" do
-        op1 = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, "json_schema_version" => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
-        op2 = new_primary_indexing_operation({"type" => "Widget", "id" => "2", "version" => 5, "json_schema_version" => 2, "record" => {"id" => "2", "dayOfWeek" => "TUE", "created_at" => "2024-09-15T13:30:12Z", "workspace_id" => "ws-2"}})
+        op1 = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, SCHEMA_VERSION_KEY => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
+        op2 = new_primary_indexing_operation({"type" => "Widget", "id" => "2", "version" => 5, SCHEMA_VERSION_KEY => 2, "record" => {"id" => "2", "dayOfWeek" => "TUE", "created_at" => "2024-09-15T13:30:12Z", "workspace_id" => "ws-2"}})
         operations = [op1, op2]
 
         results = warehouse_dumper.bulk(operations)
 
-        # Verify S3 uploads - should have 2 files (one for json_schema_version 1, one for json_schema_version 2)
+        # Verify S3 uploads - should have 2 files (one for schema version 1, one for schema version 2)
         expect(s3_client.api_requests.map { |req| req[:operation_name] }).to eq [:put_object, :put_object]
 
-        # Verify first file (json_schema_version 1)
+        # Verify first file (schema version 1)
         params1 = s3_client.api_requests[0].fetch(:params)
         expect(params1[:bucket]).to eq s3_bucket_name
         expect(params1[:key]).to match %r{Data0001/Widget/v1/2024-09-15/[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}\.jsonl\.gz}
@@ -61,7 +77,7 @@ class WarehouseLambda
         lines1 = jsonl_content1.split("\n")
         expect(lines1.size).to eq 1
 
-        # Verify second file (json_schema_version 2)
+        # Verify second file (schema version 2)
         params2 = s3_client.api_requests[1].fetch(:params)
         expect(params2[:key]).to match %r{Data0001/Widget/v2/2024-09-15/[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}\.jsonl\.gz}
 
@@ -99,8 +115,8 @@ class WarehouseLambda
       end
 
       it "writes operations of different types to separate S3 files" do
-        widget_op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, "json_schema_version" => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
-        component_op = new_primary_indexing_operation({"type" => "Component", "id" => "c1", "version" => 2, "json_schema_version" => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}})
+        widget_op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, SCHEMA_VERSION_KEY => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
+        component_op = new_primary_indexing_operation({"type" => "Component", "id" => "c1", "version" => 2, SCHEMA_VERSION_KEY => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}})
         operations = [widget_op, component_op]
 
         warehouse_dumper.bulk(operations)
@@ -112,9 +128,25 @@ class WarehouseLambda
         expect(keys[1]).to match %r{Data0001/Component/v1/2024-09-15/}
       end
 
+      it "uses the `#{WarehouseDumper::UNVERSIONED_S3_KEY_SEGMENT}` segment for an event that carries no schema version" do
+        op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
+
+        warehouse_dumper.bulk([op])
+
+        key = s3_client.api_requests.first.fetch(:params).fetch(:key)
+
+        expect(key).to start_with("Data0001/Widget/#{WarehouseDumper::UNVERSIONED_S3_KEY_SEGMENT}/2024-09-15/")
+        # The segment count must match a versioned key, so that a reader that splits the key keeps working.
+        expect(key.split("/").size).to eq(5)
+
+        expect(logged_jsons_of_type(WarehouseDumper::LOG_MSG_DUMPED_FILE)).to match [
+          a_hash_including({SCHEMA_VERSION_KEY => nil, JSON_SCHEMA_VERSION_KEY => nil})
+        ]
+      end
+
       it "logs structured information about received batch and dumped files" do
-        widget_op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, "json_schema_version" => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
-        component_op = new_primary_indexing_operation({"type" => "Component", "id" => "c1", "version" => 2, "json_schema_version" => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}})
+        widget_op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, SCHEMA_VERSION_KEY => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
+        component_op = new_primary_indexing_operation({"type" => "Component", "id" => "c1", "version" => 2, SCHEMA_VERSION_KEY => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}})
         operations = [widget_op, component_op]
 
         warehouse_dumper.bulk(operations)
@@ -127,13 +159,16 @@ class WarehouseLambda
           a_hash_including({
             "s3_bucket" => s3_bucket_name,
             "type" => "Widget",
-            "json_schema_version" => 1,
+            SCHEMA_VERSION_KEY => 1,
+            # Deprecated alias, kept for existing dashboards and monitors.
+            JSON_SCHEMA_VERSION_KEY => 1,
             "record_count" => 1
           }),
           a_hash_including({
             "s3_bucket" => s3_bucket_name,
             "type" => "Component",
-            "json_schema_version" => 1,
+            SCHEMA_VERSION_KEY => 1,
+            JSON_SCHEMA_VERSION_KEY => 1,
             "record_count" => 1
           })
         ]
@@ -183,7 +218,7 @@ class WarehouseLambda
           "type" => "Widget",
           "id" => "1",
           "version" => 3,
-          "json_schema_version" => 1,
+          SCHEMA_VERSION_KEY => 1,
           "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}
         })
 
diff --git a/elasticgraph/lib/elastic_graph/project_template/lib/app_name/shared_factories.rb b/elasticgraph/lib/elastic_graph/project_template/lib/app_name/shared_factories.rb
index 765d2a618..7c991e8c0 100644
--- a/elasticgraph/lib/elastic_graph/project_template/lib/app_name/shared_factories.rb
+++ b/elasticgraph/lib/elastic_graph/project_template/lib/app_name/shared_factories.rb
@@ -20,7 +20,9 @@
 
   factory :indexed_type_base, parent: :hash_base, traits: [:uuid_id, :versioned] do
     __typename { raise NotImplementedError, "You must supply __typename" }
-    __json_schema_version do
+    # The JSON schema version identifies the schema of each generated record. The indexing event
+    # uses the ingestion-format-neutral `schema_version` key, so name the factory attribute to match.
+    __schema_version do
       current_json_schema_version ||= begin
         json_schema_file = File.expand_path("../../config/schema/artifacts/json_schemas.yaml", __dir__)
         YAML.safe_load_file(json_schema_file).fetch("json_schema_version")
diff --git a/spec_support/lib/elastic_graph/spec_support/factories.rb b/spec_support/lib/elastic_graph/spec_support/factories.rb
index 9e54a399b..5d345d9a4 100644
--- a/spec_support/lib/elastic_graph/spec_support/factories.rb
+++ b/spec_support/lib/elastic_graph/spec_support/factories.rb
@@ -24,7 +24,7 @@ module HashAsEmbedded
         # Strips indexed-type-only keys, converting an indexed factory hash
         # to one suitable for embedding in another record.
         def as_embedded
-          except(:__version, :__json_schema_version)
+          except(:__version, :__schema_version)
         end
       end
     end
@@ -94,7 +94,7 @@ module ElasticGraphSpecSupport
     # For tests that really care about the version, they override it to control this more tightly.
     __version { version_counter += 1 }
     __typename { raise NotImplementedError, "You must supply __typename." }
-    __json_schema_version { 1 }
+    __schema_version { 1 }
     id { Faker::Alphanumeric.alpha(number: 20) }
   end