fix!: fence aggregate replay by event schema version

README.md

@@ -1054,13 +1054,27 @@ fn upcast_initialized_v1_v2((id, task): InitV1) -> InitV2 {
 }
 
 aggregate!(Todo, entity {
-    "initialized"(id, task, priority) => initialize,
+    "initialized"(id, task, priority), version = 2 => initialize,
     "completed"() => complete(),
 } upcasters [
     ("initialized", 1 => 2, InitV1 => InitV2, upcast_initialized_v1_v2),
 ]);
 ```
 
+Replay requires each event's schema version to match its registered handler
+after upcasting. `#[sourced]` reads this version from `#[event(..., version = N)]`;
+`aggregate!` declares it on the registration as above and must match the
+corresponding `#[digest(..., version = N)]`. Omitted versions default to 1.
+Older events need an explicit upcaster chain to the registered version; future
+versions are rejected. Matching payload layouts do not bypass this check.
+The check runs before payload decoding or handler invocation, including when
+loading only the event tail after a snapshot in a native repository or cell.
+
+**Breaking change:** an `aggregate!` registration for a versioned handler must
+now declare its current version. Histories with unsupported versions fail replay
+instead of being interpreted using the current payload shape. Stored events are
+unchanged; add the appropriate upcasters to read supported older versions.
+
 ## Event Metadata
 
 Metadata lets you attach cross-cutting context — correlation IDs, causation IDs, user context, trace spans — to events without changing your domain model.

distributed_macros/src/aggregate.rs

@@ -11,17 +11,18 @@ use syn::{
 ///
 /// Both entry points produce a byte-identical impl: same associated
 /// `ReplayError = String`, same `entity`/`entity_mut`/`replay_event` bodies, and
-/// the same optional `aggregate_type` and upcasters methods. Only the replay
-/// match arms differ in how they are built upstream, so this helper takes them
-/// (already rendered) along with the type name and entity field. Keeping one
-/// emitter prevents the replay semantics of the two macros from drifting.
+/// the same optional `aggregate_type` and upcasters methods. The version and
+/// replay match arms are built upstream, so this helper takes them (already
+/// rendered) along with the type name and entity field. Keeping the version
+/// fence here prevents the replay semantics of the two macros from drifting.
 ///
 /// It emits only the `impl` block; callers still place `#upcaster_wrappers`
 /// (the free upcaster fns) where they already do.
 pub(crate) fn aggregate_impl_tokens(
     type_name: &Ident,
     entity_field: &Ident,
     aggregate_type_method: &Option<TokenStream2>,
+    version_arms: &[TokenStream2],
     replay_arms: &[TokenStream2],
     upcasters_method: &TokenStream2,
 ) -> TokenStream2 {
@@ -43,6 +44,18 @@ pub(crate) fn aggregate_impl_tokens(
                 &mut self,
                 event: &distributed::EventRecord,
             ) -> Result<(), Self::ReplayError> {
+                // Hydration runs upcasters first. Payload compatibility alone
+                // cannot establish that an event has the handler's semantics.
+                let expected_version: u64 = match event.event_name.as_str() {
+                    #(#version_arms)*
+                    _ => return Err(format!("Unknown event: {}", event.event_name)),
+                };
+                if event.event_version != expected_version {
+                    return Err(format!(
+                        "Unsupported event version for {}: expected {}, got {}",
+                        event.event_name, expected_version, event.event_version,
+                    ));
+                }
                 match event.event_name.as_str() {
                     #(#replay_arms)*
                     _ => return Err(format!("Unknown event: {}", event.event_name)),
@@ -178,6 +191,15 @@ pub(crate) fn expand_aggregate(input: TokenStream2) -> syn::Result<TokenStream2>
 
     let agg_name = &input.agg_name;
     let entity_field = &input.entity_field;
+    let version_arms: Vec<_> = input
+        .events
+        .iter()
+        .map(|event| {
+            let name = &event.event_name;
+            let version = &event.version;
+            quote! { #name => #version, }
+        })
+        .collect();
 
     // Generate replay match arms - deserialize and call method directly
     let replay_arms: Vec<_> = input
@@ -249,6 +271,7 @@ pub(crate) fn expand_aggregate(input: TokenStream2) -> syn::Result<TokenStream2>
         agg_name,
         entity_field,
         &aggregate_type_method,
+        &version_arms,
         &replay_arms,
         &upcasters_method,
     );
@@ -327,6 +350,7 @@ struct AggregateInput {
 
 struct EventDef {
     event_name: LitStr,
+    version: syn::LitInt,
     args: Vec<Ident>,
     method_name: Ident,
     method_args: Option<Vec<Ident>>, // None = use event args, Some([]) = no args, Some([x,y]) = specific args
@@ -382,6 +406,19 @@ impl Parse for AggregateInput {
                 args_content.parse_terminated(Ident::parse, Token![,])?;
             let args: Vec<Ident> = args.into_iter().collect();
 
+            // `"renamed"(name), version = 2 => rename`; omitted versions
+            // have the same v1 default as #[digest] and #[event].
+            let version = if content.peek(Token![,]) {
+                content.parse::<Token![,]>()?;
+                let keyword: Ident = content.parse()?;
+                if keyword != "version" {
+                    return Err(syn::Error::new(keyword.span(), "expected `version`"));
+                }
+                content.parse::<Token![=]>()?;
+                content.parse::<syn::LitInt>()?
+            } else {
+                syn::LitInt::new("1", event_name.span())
+            };
             content.parse::<Token![=>]>()?;
             let method_name: Ident = content.parse()?;
 
@@ -398,6 +435,7 @@ impl Parse for AggregateInput {
 
             events.push(EventDef {
                 event_name,
+                version,
                 args,
                 method_name,
                 method_args,

distributed_macros/src/sourced.rs

@@ -212,6 +212,7 @@ fn find_and_remove_event_attr(
 
 struct EventMethodInfo {
     event_name: LitStr,
+    version: syn::LitInt,
     method_name: Ident,
     params: Vec<(Ident, syn::Type)>,
     /// Present when this recorder has `domain` and therefore a generated
@@ -833,6 +834,7 @@ pub(crate) fn expand_sourced(attr: TokenStream2, item: TokenStream2) -> syn::Res
                     };
 
                     event_methods.push(EventMethodInfo {
+                        version: event_version(event_attr.version.as_ref()),
                         event_name: event_attr.event_name,
                         method_name: method.sig.ident.clone(),
                         params,
@@ -977,6 +979,14 @@ pub(crate) fn expand_sourced(attr: TokenStream2, item: TokenStream2) -> syn::Res
 
     // Generate impl Aggregate
     let entity_field = &args.entity_field;
+    let version_arms: Vec<_> = event_methods
+        .iter()
+        .map(|event| {
+            let name = &event.event_name;
+            let version = &event.version;
+            quote! { #name => #version, }
+        })
+        .collect();
     let replay_arms: Vec<_> = event_methods
         .iter()
         .map(|e| {
@@ -1023,6 +1033,7 @@ pub(crate) fn expand_sourced(attr: TokenStream2, item: TokenStream2) -> syn::Res
         &struct_name,
         entity_field,
         &aggregate_type_method,
+        &version_arms,
         &replay_arms,
         &upcasters_method,
     );