feat(physics): port retail complete object frame pipeline

Restore the named-retail object update order across local, remote, static, projectile, animation, shadow, teleport, and effect lifetimes. Separate authoritative root commits from spatial rebucketing, preserve per-owner hook/FIFO ordering, and remove update-path allocations with exact lifecycle and residency gates.

Add deterministic conformance, adversarial lifetime, GUID-reuse, pending-cell, quaternion, timestamp, and allocation coverage. Release build is warning-free and all 6,446 tests pass with five intentional skips; retail, architecture, and adversarial reviews are clean.

Co-authored-by: OpenAI Codex <codex@openai.com>

src/AcDream.App/World/LiveEntityPresentationController.cs

@@ -31,6 +31,7 @@ public sealed class LiveEntityPresentationController : IDisposable
     private readonly Dictionary<uint, ushort> _readyGenerationByGuid = new();
     private readonly Dictionary<uint, ushort> _suspendedShadowGenerationByGuid = new();
     private readonly Dictionary<uint, ushort> _activePlacementGenerationByGuid = new();
+    private readonly HashSet<LiveEntityRecord> _drainingRecords = new();
     private bool _disposed;
 
     public LiveEntityPresentationController(
@@ -69,8 +70,20 @@ public sealed class LiveEntityPresentationController : IDisposable
         }
 
         _readyGenerationByGuid[serverGuid] = record.Generation;
-        ApplyPendingTransitions(record);
-        return true;
+        if (!ApplyPendingTransitions(record)
+            || record.WorldEntity is not { } entity
+            || !IsCurrent(record, entity))
+        {
+            return false;
+        }
+
+        // An object can materialize into a pending landblock before collision
+        // registration and before this ready barrier opens. Its initial false
+        // visibility edge therefore had nothing to suspend. Reconcile here,
+        // after every create-time owner exists, so the retained registration
+        // cannot remain active offscreen and hydration has a restore marker.
+        SuspendOrdinaryShadowOutsideProjection(record, entity);
+        return IsCurrent(record, entity);
     }
 
     /// <summary>Drains a newly accepted SetState when this incarnation is ready.</summary>
@@ -83,8 +96,7 @@ public sealed class LiveEntityPresentationController : IDisposable
             return false;
         }
 
-        ApplyPendingTransitions(record);
-        return true;
+        return ApplyPendingTransitions(record);
     }
 
     /// <summary>
@@ -188,6 +200,7 @@ public sealed class LiveEntityPresentationController : IDisposable
         {
             _activePlacementGenerationByGuid.Remove(record.ServerGuid);
         }
+        _drainingRecords.Remove(record);
     }
 
     public void Clear()
@@ -195,6 +208,7 @@ public sealed class LiveEntityPresentationController : IDisposable
         _readyGenerationByGuid.Clear();
         _suspendedShadowGenerationByGuid.Clear();
         _activePlacementGenerationByGuid.Clear();
+        _drainingRecords.Clear();
     }
 
     public void Dispose()
@@ -206,46 +220,83 @@ public sealed class LiveEntityPresentationController : IDisposable
         Clear();
     }
 
-    private void ApplyPendingTransitions(LiveEntityRecord record)
+    private bool ApplyPendingTransitions(LiveEntityRecord record)
     {
-        while (record.TryDequeueStateTransition(out RetailPhysicsStateTransition transition))
+        // State callbacks can synchronously accept another SetState. Retail's
+        // update thread completes the current transition before the next FIFO
+        // entry; a recursive drain would interleave Visible halfway through
+        // Hidden and then let the older side effects win last.
+        if (!_drainingRecords.Add(record))
+            return IsCurrent(record, record.WorldEntity);
+
+        try
         {
-            if (record.WorldEntity is not { } entity)
-                return;
-
-            // set_state writes the final state before any PartArray/cell side
-            // effect. Retained shadow registrations must see those same bits
-            // even while their cell rows are suspended.
-            _shadows.UpdatePhysicsState(entity.Id, (uint)transition.FinalState);
-
-            switch (transition.HiddenTransition)
+            while (record.TryDequeueStateTransition(out RetailPhysicsStateTransition transition))
             {
-                case RetailHiddenTransition.BecameHidden:
-                    _playTyped(entity.Id, HiddenScriptType, 1f);
-                    _setDirectChildrenNoDraw(record.ServerGuid, true);
-                    _shadows.Suspend(entity.Id);
-                    if (!IsPlacementActive(record))
-                        _suspendedShadowGenerationByGuid[record.ServerGuid] = record.Generation;
-                    // Retail CPhysicsObj::set_hidden @ 0x00514C60 calls
-                    // CPartArray::HandleEnterWorld after hiding the object
-                    // from its cell. Despite the name, this is the motion
-                    // timeline boundary: it strips link animations and
-                    // aborts every pending completion through
-                    // MotionTableManager::HandleEnterWorld @ 0x0051BDD0.
-                    _handlePartArrayEnterWorld(entity.Id);
-                    _clearInvalidTarget(record.ServerGuid);
-                    break;
+                if (record.WorldEntity is not { } entity
+                    || !IsCurrent(record, entity))
+                {
+                    return false;
+                }
 
-                case RetailHiddenTransition.BecameVisible:
-                    _playTyped(entity.Id, UnHideScriptType, 1f);
-                    _setDirectChildrenNoDraw(record.ServerGuid, false);
-                    // Retail invokes the same PartArray boundary before
-                    // CObjCell::unhide_object restores cell visibility.
-                    _handlePartArrayEnterWorld(entity.Id);
-                    if (!IsPlacementActive(record) && RestoreShadow(record, entity))
-                        _suspendedShadowGenerationByGuid.Remove(record.ServerGuid);
-                    break;
+                // set_state writes the final state before any PartArray/cell side
+                // effect. Retained shadow registrations must see those same bits
+                // even while their cell rows are suspended.
+                _shadows.UpdatePhysicsState(entity.Id, (uint)transition.FinalState);
+
+                switch (transition.HiddenTransition)
+                {
+                    case RetailHiddenTransition.BecameHidden:
+                        _playTyped(entity.Id, HiddenScriptType, 1f);
+                        if (!IsCurrent(record, entity))
+                            return false;
+                        _setDirectChildrenNoDraw(record.ServerGuid, true);
+                        if (!IsCurrent(record, entity))
+                            return false;
+                        _shadows.Suspend(entity.Id);
+                        if (!IsPlacementActive(record))
+                            _suspendedShadowGenerationByGuid[record.ServerGuid] = record.Generation;
+                        // Retail CPhysicsObj::set_hidden @ 0x00514C60 calls
+                        // CPartArray::HandleEnterWorld after hiding the object
+                        // from its cell. Despite the name, this is the motion
+                        // timeline boundary: it strips link animations and
+                        // aborts every pending completion through
+                        // MotionTableManager::HandleEnterWorld @ 0x0051BDD0.
+                        _handlePartArrayEnterWorld(entity.Id);
+                        if (!IsCurrent(record, entity))
+                            return false;
+                        _clearInvalidTarget(record.ServerGuid);
+                        if (!IsCurrent(record, entity))
+                            return false;
+                        break;
+
+                    case RetailHiddenTransition.BecameVisible:
+                        _playTyped(entity.Id, UnHideScriptType, 1f);
+                        if (!IsCurrent(record, entity))
+                            return false;
+                        _setDirectChildrenNoDraw(record.ServerGuid, false);
+                        if (!IsCurrent(record, entity))
+                            return false;
+                        // Retail invokes the same PartArray boundary before
+                        // CObjCell::unhide_object restores cell visibility.
+                        _handlePartArrayEnterWorld(entity.Id);
+                        if (!IsCurrent(record, entity))
+                            return false;
+                        bool restored = !IsPlacementActive(record)
+                            && RestoreShadow(record, entity);
+                        if (!IsCurrent(record, entity))
+                            return false;
+                        if (restored)
+                            _suspendedShadowGenerationByGuid.Remove(record.ServerGuid);
+                        break;
+                }
             }
+
+            return IsCurrent(record, record.WorldEntity);
+        }
+        finally
+        {
+            _drainingRecords.Remove(record);
         }
     }
 
@@ -273,9 +324,25 @@ public sealed class LiveEntityPresentationController : IDisposable
 
     private void OnProjectionVisibilityChanged(LiveEntityRecord record, bool visible)
     {
-        if (!visible
-            || record.WorldEntity is not { } entity
-            || (record.FinalPhysicsState & PhysicsStateFlags.Hidden) != 0
+        if (record.WorldEntity is not { } entity
+            || !IsCurrent(record, entity))
+        {
+            return;
+        }
+
+        // exit_world removes an ordinary object's collision rows on the same
+        // projection edge that suspends its object clock. Keep the retained
+        // registration so a stationary object can be restored immediately on
+        // hydration; it may have no later movement quantum to repair itself.
+        // Projectiles and active authoritative placements own their matching
+        // suspend/restore transaction in their dedicated controllers.
+        if (!visible)
+        {
+            SuspendOrdinaryShadowOutsideProjection(record, entity);
+            return;
+        }
+
+        if ((record.FinalPhysicsState & PhysicsStateFlags.Hidden) != 0
             || IsPlacementActive(record)
             || !_readyGenerationByGuid.TryGetValue(record.ServerGuid, out ushort readyGeneration)
             || readyGeneration != record.Generation
@@ -287,10 +354,40 @@ public sealed class LiveEntityPresentationController : IDisposable
             return;
         }
 
-        if (RestoreShadow(record, entity))
+        bool restored = RestoreShadow(record, entity);
+        if (!IsCurrent(record, entity))
+            return;
+        if (restored)
             _suspendedShadowGenerationByGuid.Remove(record.ServerGuid);
     }
 
+    private void SuspendOrdinaryShadowOutsideProjection(
+        LiveEntityRecord record,
+        AcDream.Core.World.WorldEntity entity)
+    {
+        if (record.IsSpatiallyVisible
+            || record.ProjectileRuntime is not null
+            || IsPlacementActive(record)
+            || !_readyGenerationByGuid.TryGetValue(
+                record.ServerGuid,
+                out ushort readyGeneration)
+            || readyGeneration != record.Generation
+            || !_shadows.Suspend(entity.Id))
+        {
+            return;
+        }
+
+        _suspendedShadowGenerationByGuid[record.ServerGuid] = record.Generation;
+    }
+
+    private bool IsCurrent(
+        LiveEntityRecord record,
+        AcDream.Core.World.WorldEntity? entity) =>
+        entity is not null
+        && _liveEntities.TryGetRecord(record.ServerGuid, out LiveEntityRecord current)
+        && ReferenceEquals(current, record)
+        && ReferenceEquals(current.WorldEntity, entity);
+
     private bool IsPlacementActive(LiveEntityRecord record) =>
         _activePlacementGenerationByGuid.TryGetValue(
             record.ServerGuid,

src/AcDream.App/World/LiveEntityRuntime.cs

@@ -48,6 +48,7 @@ public interface ILiveEntityRemotePlacementRuntime :
     Vector3 LastServerPosition { get; set; }
     double LastServerPositionTime { get; set; }
     Vector3 LastShadowSyncPosition { get; set; }
+    Quaternion LastShadowSyncOrientation { get; set; }
     void HitGround();
     void LeaveGround();
 }
@@ -111,6 +112,13 @@ public sealed class LiveEntityRecord
         ServerGuid = snapshot.Guid;
         Snapshot = snapshot;
         RefreshDerivedState();
+        PositionAuthorityVersion = snapshot.Position is null ? 0UL : 1UL;
+        VectorAuthorityVersion = snapshot.Physics is null ? 0UL : 1UL;
+        VelocityAuthorityVersion = snapshot.Position is null
+            && snapshot.Physics is null
+                ? 0UL
+                : 1UL;
+        MovementAuthorityVersion = 1UL;
         FinalPhysicsState = RetailPhysicsStateTransitions.ConstructorState;
         ApplyRawPhysicsState(RawPhysicsState);
     }
@@ -124,7 +132,29 @@ public sealed class LiveEntityRecord
     public uint CanonicalLandblockId { get; internal set; }
     public uint RawPhysicsState { get; internal set; }
     public PhysicsStateFlags FinalPhysicsState { get; internal set; }
+    /// <summary>
+    /// Incarnation-stable retail <c>CPhysicsObj::update_time</c> owner. It
+    /// survives animation and MovementManager attachment/replacement, spatial
+    /// rebucketing, and appearance rehydration, and dies with this record.
+    /// </summary>
+    public RetailObjectQuantumClock ObjectClock { get; } = new();
+    /// <summary>
+    /// Changes whenever this record leaves or re-enters the ordinary-object
+    /// workset. Unlike <see cref="ProjectionMutationVersion"/>, loaded-to-loaded
+    /// rebucketing does not change this token. A scheduler can therefore stop a
+    /// catch-up batch that crossed a leave/enter boundary without rejecting an
+    /// ordinary same-frame cell move.
+    /// </summary>
+    internal ulong ObjectClockEpoch { get; private set; }
+    /// <summary>
+    /// True after this incarnation has successfully constructed its retail
+    /// <c>CPartArray</c>. This is an object-lifetime fact, not an animation-
+    /// runtime proxy: static poses and objects without a MotionTable still
+    /// own a PartArray and participate in the 96-unit Active gate.
+    /// </summary>
+    public bool HasPartArray { get; internal set; }
     public PhysicsBody? PhysicsBody { get; internal set; }
+    internal bool PhysicsBodyAcquisitionInProgress { get; set; }
     public ILiveEntityAnimationRuntime? AnimationRuntime { get; internal set; }
     public ILiveEntityRemoteMotionRuntime? RemoteMotionRuntime { get; internal set; }
     internal bool RequiresRemotePlacementRuntime { get; set; }
@@ -134,6 +164,16 @@ public sealed class LiveEntityRecord
     public bool IsSpatiallyProjected { get; internal set; }
     public bool IsSpatiallyVisible { get; internal set; }
     internal ulong ProjectionMutationVersion { get; set; }
+    /// <summary>
+    /// Monotonic App operation tokens for accepted wire authority. They let a
+    /// handler detect a newer same-incarnation packet delivered re-entrantly
+    /// while an older packet is crossing presentation or spatial callbacks.
+    /// </summary>
+    internal ulong PositionAuthorityVersion { get; private set; }
+    internal ulong StateAuthorityVersion { get; private set; }
+    internal ulong VectorAuthorityVersion { get; private set; }
+    internal ulong VelocityAuthorityVersion { get; private set; }
+    internal ulong MovementAuthorityVersion { get; private set; }
     public bool WorldSpawnPublished { get; internal set; }
     public LiveEntityProjectionKind ProjectionKind { get; internal set; }
     internal bool RuntimeComponentsTeardownCompleted { get; set; }
@@ -145,8 +185,21 @@ public sealed class LiveEntityRecord
     internal bool TryDequeueStateTransition(out RetailPhysicsStateTransition transition) =>
         _pendingStateTransitions.TryDequeue(out transition);
 
+    internal void SuspendObjectClock()
+    {
+        ObjectClock.Deactivate();
+        ObjectClockEpoch++;
+    }
+
+    internal void ResetObjectClockForEnterWorld(bool isStatic)
+    {
+        ObjectClock.ResetForEnterWorld(isStatic);
+        ObjectClockEpoch++;
+    }
+
     internal RetailPhysicsStateTransition ApplyRawPhysicsState(uint rawState)
     {
+        StateAuthorityVersion++;
         RawPhysicsState = rawState;
         RetailPhysicsStateTransition transition = RetailPhysicsStateTransitions.Apply(
             FinalPhysicsState,
@@ -158,6 +211,33 @@ public sealed class LiveEntityRecord
         return transition;
     }
 
+    internal void AdvancePositionAuthority()
+    {
+        PositionAuthorityVersion++;
+        VelocityAuthorityVersion++;
+    }
+
+    internal void AdvanceVectorAuthority()
+    {
+        VectorAuthorityVersion++;
+        VelocityAuthorityVersion++;
+    }
+
+    internal void AdvanceMovementAuthority()
+    {
+        MovementAuthorityVersion++;
+        VelocityAuthorityVersion++;
+    }
+
+    internal void AdvanceCreateAuthority()
+    {
+        PositionAuthorityVersion++;
+        StateAuthorityVersion++;
+        VectorAuthorityVersion++;
+        VelocityAuthorityVersion++;
+        MovementAuthorityVersion++;
+    }
+
     internal void SetChildNoDraw(bool noDraw)
     {
         FinalPhysicsState = noDraw
@@ -241,6 +321,10 @@ public sealed class LiveEntityRuntime
     private readonly Dictionary<uint, ILiveEntityAnimationRuntime> _spatialAnimationsByLocalId = new();
     private readonly Dictionary<uint, ILiveEntityRemoteMotionRuntime> _spatialRemoteMotionByGuid = new();
     private readonly Dictionary<uint, ILiveEntityProjectileRuntime> _spatialProjectilesByGuid = new();
+    // Retail CPhysics::UseTime walks the ordinary object table, not a render-
+    // animation component table. This index is the loaded/cell-backed root
+    // workset; component-specific indexes remain lookup accelerators only.
+    private readonly Dictionary<uint, LiveEntityRecord> _spatialRootObjectsByGuid = new();
     private bool _isClearing;
     private bool _sessionClearPendingFinalization;
     private bool _isRegisteringResources;
@@ -297,6 +381,8 @@ public sealed class LiveEntityRuntime
         _spatialRemoteMotionByGuid;
     internal IReadOnlyDictionary<uint, ILiveEntityProjectileRuntime> SpatialProjectileRuntimes =>
         _spatialProjectilesByGuid;
+    internal IReadOnlyDictionary<uint, LiveEntityRecord> SpatialRootObjects =>
+        _spatialRootObjectsByGuid;
     public ParentAttachmentState ParentAttachments { get; } = new();
 
     /// <summary>
@@ -327,6 +413,7 @@ public sealed class LiveEntityRuntime
             {
                 retained.Snapshot = result.Snapshot;
                 retained.RefreshDerivedState();
+                retained.AdvanceCreateAuthority();
                 RefreshSpatialRuntimeIndexes(retained);
                 return new LiveEntityRegistrationResult(result, retained, false, false);
             }
@@ -510,8 +597,18 @@ public sealed class LiveEntityRuntime
             record.WorldEntity!.IsAncestorDrawVisible = true;
         }
         RefreshPresentation(record);
-        RebucketLiveEntity(serverGuid, fullCellId);
-        return record.WorldEntity;
+        WorldEntity materialized = record.WorldEntity!;
+        if (!RebucketLiveEntity(serverGuid, fullCellId)
+            || !_recordsByGuid.TryGetValue(serverGuid, out LiveEntityRecord? current)
+            || !ReferenceEquals(current, record)
+            || !ReferenceEquals(current.WorldEntity, materialized))
+        {
+            // Visibility observers may replace this GUID while rebucketing.
+            // A failed teardown can retain the displaced entity as a retry
+            // tombstone; it is not the result of this materialization.
+            return null;
+        }
+        return materialized;
     }
 
     /// <summary>Changes spatial buckets only. No logical resource callbacks run.</summary>
@@ -523,6 +620,10 @@ public sealed class LiveEntityRuntime
 
         bool wasProjected = record.IsSpatiallyProjected;
         bool wasVisible = record.IsSpatiallyVisible;
+        bool wasOrdinaryRoot = _spatialRootObjectsByGuid.TryGetValue(
+                serverGuid,
+                out LiveEntityRecord? indexedRoot)
+            && ReferenceEquals(indexedRoot, record);
         ulong projectionOperation = ++record.ProjectionMutationVersion;
         // GpuWorldState reports an intermediate false/true pair while moving
         // between two loaded buckets. Suppress those implementation details
@@ -570,6 +671,38 @@ public sealed class LiveEntityRuntime
         record.CanonicalLandblockId = spatialCellOrLandblockId == 0
             ? 0u
             : (spatialCellOrLandblockId & 0xFFFF0000u) | 0xFFFFu;
+        bool isOrdinaryRoot = record.ProjectionKind is LiveEntityProjectionKind.World
+            && record.IsSpatiallyProjected
+            && record.IsSpatiallyVisible
+            && record.FullCellId != 0;
+        // prepare_to_enter_world (0x00511FA0) writes update_time = cur_time.
+        // Only a root-workset membership edge rebases the clock. Ordinary
+        // loaded-to-loaded movement preserves it; parented/attached objects
+        // take retail update_object's parent early-out and remain suspended.
+        if (!wasProjected && !isOrdinaryRoot)
+        {
+            record.SuspendObjectClock();
+            SynchronizePhysicsBodyActiveState(record);
+        }
+        else if (wasOrdinaryRoot != isOrdinaryRoot)
+        {
+            if (isOrdinaryRoot)
+            {
+                // InitObjectBegin and every leave-visibility edge are inactive
+                // before retail prepare_to_enter_world. Preserve that
+                // prerequisite for a first materialization as well.
+                if (!wasProjected)
+                    record.ObjectClock.Deactivate();
+                record.ResetObjectClockForEnterWorld(
+                    (record.FinalPhysicsState & PhysicsStateFlags.Static) != 0);
+                SynchronizePhysicsBodyActiveState(record);
+            }
+            else
+            {
+                record.SuspendObjectClock();
+                SynchronizePhysicsBodyActiveState(record);
+            }
+        }
         RefreshSpatialRuntimeIndexes(record);
         Exception? runtimeNotificationFailure = null;
         if (!wasProjected || wasVisible != visible)
@@ -583,6 +716,18 @@ public sealed class LiveEntityRuntime
                 runtimeNotificationFailure = error;
             }
         }
+        // Final observers are intentionally re-entrant. A callback can delete
+        // and recreate this GUID or start a newer rebucket on the same record;
+        // the outer caller must not continue with captures from that displaced
+        // projection operation.
+        if (!IsCurrentProjectionOperation(serverGuid, record, projectionOperation))
+        {
+            ThrowAfterCommittedProjectionChange(
+                serverGuid,
+                spatialNotificationFailure,
+                runtimeNotificationFailure);
+            return false;
+        }
         ThrowAfterCommittedProjectionChange(
             serverGuid,
             spatialNotificationFailure,
@@ -600,6 +745,11 @@ public sealed class LiveEntityRuntime
             || record.WorldEntity is null)
             return false;
 
+        bool wasOrdinaryRoot = _spatialRootObjectsByGuid.TryGetValue(
+            serverGuid,
+            out LiveEntityRecord? rootRecord)
+            && ReferenceEquals(rootRecord, record);
+        ulong objectClockEpoch = record.ObjectClockEpoch;
         ulong projectionOperation = ++record.ProjectionMutationVersion;
         Exception? spatialNotificationFailure = null;
         try
@@ -631,6 +781,14 @@ public sealed class LiveEntityRuntime
         _visibleWorldEntitiesByGuid.Remove(serverGuid);
         record.IsSpatiallyProjected = false;
         record.IsSpatiallyVisible = false;
+        // GpuWorldState normally publishes the leave edge synchronously and
+        // OnSpatialVisibilityChanged suspends this exact clock there. During a
+        // reentrant notification the edge is deferred, so commit the missing
+        // suspension here. One root-workset leave is one epoch edge -- never
+        // advance it once in the observer and again in this owner.
+        if (wasOrdinaryRoot && record.ObjectClockEpoch == objectClockEpoch)
+            record.SuspendObjectClock();
+        SynchronizePhysicsBodyActiveState(record);
         RefreshSpatialRuntimeIndexes(record);
         Exception? runtimeNotificationFailure = null;
         if (spatialEdgeWasDeferred)
@@ -644,6 +802,18 @@ public sealed class LiveEntityRuntime
                 runtimeNotificationFailure = error;
             }
         }
+        // The manually published deferred edge is an arbitrary callback
+        // boundary just like Rebucket's final visibility edge. It may
+        // reproject this record or replace the GUID; the displaced outer
+        // withdrawal must not report itself as the current operation.
+        if (!IsCurrentProjectionOperation(serverGuid, record, projectionOperation))
+        {
+            ThrowAfterCommittedProjectionChange(
+                serverGuid,
+                spatialNotificationFailure,
+                runtimeNotificationFailure);
+            return false;
+        }
         ThrowAfterCommittedProjectionChange(
             serverGuid,
             spatialNotificationFailure,
@@ -845,11 +1015,70 @@ public sealed class LiveEntityRuntime
         return true;
     }
 
+    /// <summary>
+    /// Acquires the one retail <c>CPhysicsObj</c> body for this exact live
+    /// incarnation. Static animation may need the body before a MovementManager
+    /// or projectile component exists; later components must adopt this same
+    /// instance instead of replaying CreateObject vector initialization.
+    /// </summary>
+    internal PhysicsBody GetOrCreatePhysicsBody(
+        uint serverGuid,
+        Func<LiveEntityRecord, PhysicsBody> factory)
+    {
+        ArgumentNullException.ThrowIfNull(factory);
+        if (!_recordsByGuid.TryGetValue(serverGuid, out LiveEntityRecord? record)
+            || record.WorldEntity is null)
+        {
+            throw new InvalidOperationException(
+                $"Cannot acquire physics body before live entity 0x{serverGuid:X8} is materialized.");
+        }
+        if (record.PhysicsBody is { } retained)
+            return retained;
+        if (record.PhysicsBodyAcquisitionInProgress)
+        {
+            throw new InvalidOperationException(
+                $"Live entity 0x{serverGuid:X8} physics-body acquisition is already in progress.");
+        }
+
+        record.PhysicsBodyAcquisitionInProgress = true;
+        try
+        {
+            PhysicsBody candidate = factory(record)
+                ?? throw new InvalidOperationException("Physics-body factory returned null.");
+            if (!_recordsByGuid.TryGetValue(serverGuid, out LiveEntityRecord? current)
+                || !ReferenceEquals(current, record)
+                || current.WorldEntity is null)
+            {
+                throw new InvalidOperationException(
+                    $"Live entity 0x{serverGuid:X8} changed incarnation during physics-body acquisition.");
+            }
+            if (record.PhysicsBody is { } concurrentlyBound)
+            {
+                if (ReferenceEquals(concurrentlyBound, candidate))
+                    return concurrentlyBound;
+                throw new InvalidOperationException(
+                    $"Live entity 0x{serverGuid:X8} acquired two physics bodies within one incarnation.");
+            }
+
+            record.PhysicsBody = candidate;
+            candidate.State = record.FinalPhysicsState;
+            SynchronizePhysicsBodyActiveState(record);
+            return candidate;
+        }
+        finally
+        {
+            record.PhysicsBodyAcquisitionInProgress = false;
+        }
+    }
+
     public void SetRemoteMotionRuntime(uint serverGuid, ILiveEntityRemoteMotionRuntime runtime)
     {
         ArgumentNullException.ThrowIfNull(runtime);
         if (!_recordsByGuid.TryGetValue(serverGuid, out LiveEntityRecord? record))
             throw new InvalidOperationException($"Cannot bind remote motion before live entity 0x{serverGuid:X8} exists.");
+        if (record.PhysicsBodyAcquisitionInProgress && record.PhysicsBody is null)
+            throw new InvalidOperationException(
+                $"Live entity 0x{serverGuid:X8} cannot bind remote motion during physics-body acquisition.");
         PhysicsBody candidateBody = runtime.Body;
         if (record.PhysicsBody is { } canonicalBody
             && !ReferenceEquals(canonicalBody, candidateBody))
@@ -868,6 +1097,7 @@ public sealed class LiveEntityRuntime
         record.RemoteMotionRuntime = runtime;
         record.PhysicsBody = candidateBody;
         candidateBody.State = record.FinalPhysicsState;
+        SynchronizePhysicsBodyActiveState(record);
         if (runtime is ILiveEntityCanonicalCellConsumer cellConsumer)
         {
             // Capture the incarnation, not only its GUID. A callback retained
@@ -919,6 +1149,9 @@ public sealed class LiveEntityRuntime
             throw new InvalidOperationException(
                 $"Cannot bind projectile physics before live entity 0x{serverGuid:X8} is materialized.");
         }
+        if (record.PhysicsBodyAcquisitionInProgress && record.PhysicsBody is null)
+            throw new InvalidOperationException(
+                $"Live entity 0x{serverGuid:X8} cannot bind projectile physics during physics-body acquisition.");
         PhysicsBody candidateBody = runtime.Body;
         if (record.PhysicsBody is { } canonicalBody
             && !ReferenceEquals(canonicalBody, candidateBody))
@@ -930,6 +1163,7 @@ public sealed class LiveEntityRuntime
         record.ProjectileRuntime = runtime;
         record.PhysicsBody = candidateBody;
         candidateBody.State = record.FinalPhysicsState;
+        SynchronizePhysicsBodyActiveState(record);
         RefreshSpatialRuntimeIndexes(record);
     }
 
@@ -999,6 +1233,8 @@ public sealed class LiveEntityRuntime
         if (applied)
         {
             RefreshRecord(update.Guid, accepted);
+            if (_recordsByGuid.TryGetValue(update.Guid, out LiveEntityRecord? record))
+                record.AdvancePositionAuthority();
             ClearWorldCell(update.Guid);
             ParentAttachments.EndChildProjection(update.Guid);
         }
@@ -1011,6 +1247,8 @@ public sealed class LiveEntityRuntime
         if (applied)
         {
             RefreshRecord(update.ChildGuid, accepted);
+            if (_recordsByGuid.TryGetValue(update.ChildGuid, out LiveEntityRecord? record))
+                record.AdvancePositionAuthority();
             ClearWorldCell(update.ChildGuid);
         }
         return applied;
@@ -1022,6 +1260,8 @@ public sealed class LiveEntityRuntime
         if (applied)
         {
             RefreshRecord(update.ChildGuid, accepted);
+            if (_recordsByGuid.TryGetValue(update.ChildGuid, out LiveEntityRecord? record))
+                record.AdvancePositionAuthority();
             ClearWorldCell(update.ChildGuid);
         }
         return applied;
@@ -1036,13 +1276,24 @@ public sealed class LiveEntityRuntime
         bool applied = _inbound.TryApplyMotion(update, retainPayload, out accepted, out timestamps);
         if (_inbound.TryGetSnapshot(update.Guid, out WorldSession.EntitySpawn snapshot))
             RefreshRecord(update.Guid, snapshot);
+        if (applied
+            && retainPayload
+            && _recordsByGuid.TryGetValue(update.Guid, out LiveEntityRecord? record))
+        {
+            record.AdvanceMovementAuthority();
+        }
         return applied;
     }
 
     public bool TryApplyVector(VectorUpdate.Parsed update, out WorldSession.EntitySpawn accepted)
     {
         bool applied = _inbound.TryApplyVector(update, out accepted);
-        if (applied) RefreshRecord(update.Guid, accepted);
+        if (applied)
+        {
+            RefreshRecord(update.Guid, accepted);
+            if (_recordsByGuid.TryGetValue(update.Guid, out LiveEntityRecord? record))
+                record.AdvanceVectorAuthority();
+        }
         return applied;
     }
 
@@ -1117,6 +1368,13 @@ public sealed class LiveEntityRuntime
             RefreshRecord(update.Guid, snapshot, refreshPosition: acceptedPosition);
             if (acceptedPosition)
             {
+                if (_recordsByGuid.TryGetValue(
+                        update.Guid,
+                        out LiveEntityRecord? acceptedRecord)
+                    && ReferenceEquals(acceptedRecord, beforePosition))
+                {
+                    acceptedRecord.AdvancePositionAuthority();
+                }
                 ParentAttachments.EndChildProjection(update.Guid);
             }
         }
@@ -1136,12 +1394,231 @@ public sealed class LiveEntityRuntime
     /// script/effect owners but clears the cell and withdraws the root
     /// projection until a fresh Position re-enters it.
     /// </summary>
+    public RetailObjectClockDisposition GetRootObjectClockDisposition(
+        uint serverGuid)
+    {
+        if (!_recordsByGuid.TryGetValue(serverGuid, out LiveEntityRecord? record)
+            || record.ProjectionKind is not LiveEntityProjectionKind.World
+            || !record.IsSpatiallyProjected
+            || !record.IsSpatiallyVisible
+            || record.FullCellId == 0)
+        {
+            return RetailObjectClockDisposition.Suspend;
+        }
+
+        return (record.FinalPhysicsState & PhysicsStateFlags.Frozen) != 0
+            ? RetailObjectClockDisposition.Suspend
+            : RetailObjectClockDisposition.Advance;
+    }
+
     public bool ShouldAdvanceRootRuntime(uint serverGuid) =>
-        _recordsByGuid.TryGetValue(serverGuid, out LiveEntityRecord? record)
-        && record.IsSpatiallyProjected
-        && record.IsSpatiallyVisible
-        && record.FullCellId != 0
-        && (record.FinalPhysicsState & PhysicsStateFlags.Frozen) == 0;
+        GetRootObjectClockDisposition(serverGuid)
+            is RetailObjectClockDisposition.Advance;
+
+    internal bool IsCurrentPositionAuthority(
+        LiveEntityRecord record,
+        ulong authorityVersion) =>
+        _recordsByGuid.TryGetValue(record.ServerGuid, out LiveEntityRecord? current)
+        && ReferenceEquals(current, record)
+        && current.PositionAuthorityVersion == authorityVersion;
+
+    internal bool IsCurrentStateAuthority(
+        LiveEntityRecord record,
+        ulong authorityVersion) =>
+        _recordsByGuid.TryGetValue(record.ServerGuid, out LiveEntityRecord? current)
+        && ReferenceEquals(current, record)
+        && current.StateAuthorityVersion == authorityVersion;
+
+    internal bool IsCurrentVectorAuthority(
+        LiveEntityRecord record,
+        ulong authorityVersion) =>
+        _recordsByGuid.TryGetValue(record.ServerGuid, out LiveEntityRecord? current)
+        && ReferenceEquals(current, record)
+        && current.VectorAuthorityVersion == authorityVersion;
+
+    internal bool IsCurrentVelocityAuthority(
+        LiveEntityRecord record,
+        ulong authorityVersion) =>
+        _recordsByGuid.TryGetValue(record.ServerGuid, out LiveEntityRecord? current)
+        && ReferenceEquals(current, record)
+        && current.VelocityAuthorityVersion == authorityVersion;
+
+    internal bool IsCurrentMovementAuthority(
+        LiveEntityRecord record,
+        ulong authorityVersion) =>
+        _recordsByGuid.TryGetValue(record.ServerGuid, out LiveEntityRecord? current)
+        && ReferenceEquals(current, record)
+        && current.MovementAuthorityVersion == authorityVersion;
+
+    /// <summary>
+    /// Copies the canonical loaded root-object workset used by retail
+    /// <c>CPhysics::UseTime</c>. The record reference is the incarnation token;
+    /// callers must revalidate it after callbacks that can delete or rebucket.
+    /// </summary>
+    internal void CopySpatialRootObjectRecordsTo(List<LiveEntityRecord> destination)
+    {
+        ArgumentNullException.ThrowIfNull(destination);
+        destination.Clear();
+        foreach ((uint serverGuid, LiveEntityRecord indexed) in _spatialRootObjectsByGuid)
+        {
+            if (_recordsByGuid.TryGetValue(serverGuid, out LiveEntityRecord? current)
+                && ReferenceEquals(current, indexed)
+                && HasSpatialRuntimeProjection(current))
+            {
+                destination.Add(current);
+            }
+        }
+    }
+
+    internal bool IsCurrentSpatialRootObject(LiveEntityRecord record) =>
+        IsCurrentRecord(record)
+        && _spatialRootObjectsByGuid.TryGetValue(record.ServerGuid, out var indexed)
+        && ReferenceEquals(indexed, record)
+        && HasSpatialRuntimeProjection(record);
+
+    internal bool IsCurrentSpatialAnimation(
+        LiveEntityRecord record,
+        ILiveEntityAnimationRuntime animation) =>
+        IsCurrentSpatialRootObject(record)
+        && ReferenceEquals(record.AnimationRuntime, animation)
+        && record.WorldEntity is { } entity
+        && _spatialAnimationsByLocalId.TryGetValue(entity.Id, out var indexed)
+        && ReferenceEquals(indexed, animation);
+
+    /// <summary>
+    /// Copies the currently spatial animation owners through the concrete
+    /// dictionary enumerator. This keeps per-frame traversal allocation-free;
+    /// exposing the workset as <see cref="IReadOnlyDictionary{TKey,TValue}"/>
+    /// would box its enumerator at a hot call site.
+    /// </summary>
+    internal void CopySpatialAnimationRuntimesTo<TAnimation>(
+        List<KeyValuePair<uint, TAnimation>> destination)
+        where TAnimation : class, ILiveEntityAnimationRuntime
+    {
+        ArgumentNullException.ThrowIfNull(destination);
+        destination.Clear();
+        foreach ((uint localId, ILiveEntityAnimationRuntime animation)
+                 in _spatialAnimationsByLocalId)
+        {
+            if (animation is TAnimation typed)
+            {
+                destination.Add(
+                    new KeyValuePair<uint, TAnimation>(localId, typed));
+            }
+        }
+    }
+
+    internal void CopySpatialAnimationLocalIdsTo(HashSet<uint> destination)
+    {
+        ArgumentNullException.ThrowIfNull(destination);
+        destination.Clear();
+        foreach (uint localId in _spatialAnimationsByLocalId.Keys)
+            destination.Add(localId);
+    }
+
+    /// <summary>
+    /// Host side of retail <c>CPhysicsObj::set_active(1)</c> for a movement
+    /// manager owned by this exact incarnation. Static is a required no-op.
+    /// </summary>
+    internal bool TryActivateOrdinaryObject(
+        uint serverGuid,
+        ILiveEntityRemoteMotionRuntime runtime)
+    {
+        ArgumentNullException.ThrowIfNull(runtime);
+        if (!_recordsByGuid.TryGetValue(serverGuid, out LiveEntityRecord? record)
+            || !ReferenceEquals(record.RemoteMotionRuntime, runtime)
+            || (record.FinalPhysicsState & PhysicsStateFlags.Static) != 0)
+        {
+            return false;
+        }
+
+        record.ObjectClock.Activate();
+        runtime.Body.TransientState |= TransientStateFlags.Active;
+        return true;
+    }
+
+    /// <summary>
+    /// Commits retail <c>CPhysicsObj::set_velocity</c> (0x005113F0) to the
+    /// canonical body for this exact incarnation. The body and the retained
+    /// object clock are one CPhysicsObj: a non-Static inactive-to-active edge
+    /// rebases both together, while Static preserves its prior Active state.
+    /// </summary>
+    internal bool TryCommitAuthoritativeVelocity(
+        LiveEntityRecord record,
+        PhysicsBody body,
+        Vector3 velocity,
+        double currentTime) =>
+        TryCommitAuthoritativeVectorCore(
+            record,
+            body,
+            velocity,
+            angularVelocity: null,
+            currentTime);
+
+    /// <summary>
+    /// Commits the paired retail VectorUpdate writes
+    /// (<c>set_velocity</c>, then <c>set_omega</c>) to one canonical body.
+    /// Validation is atomic: malformed wire vectors cannot partially mutate
+    /// the live object.
+    /// </summary>
+    internal bool TryCommitAuthoritativeVector(
+        LiveEntityRecord record,
+        PhysicsBody body,
+        Vector3 velocity,
+        Vector3 angularVelocity,
+        double currentTime) =>
+        TryCommitAuthoritativeVectorCore(
+            record,
+            body,
+            velocity,
+            angularVelocity,
+            currentTime);
+
+    private bool TryCommitAuthoritativeVectorCore(
+        LiveEntityRecord record,
+        PhysicsBody body,
+        Vector3 velocity,
+        Vector3? angularVelocity,
+        double currentTime)
+    {
+        ArgumentNullException.ThrowIfNull(record);
+        ArgumentNullException.ThrowIfNull(body);
+        if (!IsFinite(velocity)
+            || (angularVelocity is { } omega && !IsFinite(omega))
+            || !double.IsFinite(currentTime)
+            || !_recordsByGuid.TryGetValue(
+                record.ServerGuid,
+                out LiveEntityRecord? current)
+            || !ReferenceEquals(current, record)
+            || !ReferenceEquals(current.PhysicsBody, body))
+        {
+            return false;
+        }
+
+        bool wasBodyActive =
+            (body.TransientState & TransientStateFlags.Active) != 0;
+        body.set_velocity(velocity);
+        if ((current.FinalPhysicsState & PhysicsStateFlags.Static) != 0)
+        {
+            // PhysicsBody models set_velocity's leaf write and therefore sets
+            // Active unconditionally. Retail CPhysicsObj::set_active(1) is a
+            // no-op for Static, so restore the exact prior bit here.
+            if (!wasBodyActive)
+                body.TransientState &= ~TransientStateFlags.Active;
+        }
+        else
+        {
+            bool clockReactivated = current.ObjectClock.Activate();
+            // PhysicsBody.LastUpdateTime remains only the compatibility clock
+            // for older absolute-time helpers; the record clock is canonical.
+            if (clockReactivated || !wasBodyActive)
+                body.LastUpdateTime = currentTime;
+        }
+
+        if (angularVelocity is { } acceptedOmega)
+            body.Omega = acceptedOmega;
+        return true;
+    }
 
     /// <summary>
     /// Copies the currently spatial remote-motion owners into caller-owned
@@ -1393,6 +1870,19 @@ public sealed class LiveEntityRuntime
     {
         if (!_recordsByGuid.TryGetValue(guid, out LiveEntityRecord? record))
             return;
+        bool wasOrdinaryRoot = _spatialRootObjectsByGuid.TryGetValue(
+                guid,
+                out LiveEntityRecord? indexedRoot)
+            && ReferenceEquals(indexedRoot, record);
+        // Pickup/Parent is retail's leave-world edge. Suspend the canonical
+        // object clock while the record is still indexed as a root; clearing
+        // FullCellId first removes that evidence and makes the later projection
+        // withdrawal look like a duplicate non-root edge.
+        if (wasOrdinaryRoot)
+        {
+            record.SuspendObjectClock();
+            SynchronizePhysicsBodyActiveState(record);
+        }
         record.FullCellId = 0;
         record.CanonicalLandblockId = 0;
         RefreshSpatialRuntimeIndexes(record);
@@ -1404,6 +1894,7 @@ public sealed class LiveEntityRuntime
 
     private static bool HasSpatialRuntimeProjection(LiveEntityRecord record) =>
         record.WorldEntity is not null
+        && record.ProjectionKind is LiveEntityProjectionKind.World
         && record.IsSpatiallyProjected
         && record.IsSpatiallyVisible
         && record.FullCellId != 0;
@@ -1413,6 +1904,17 @@ public sealed class LiveEntityRuntime
         bool current = IsCurrentRecord(record);
         bool spatial = current && HasSpatialRuntimeProjection(record);
 
+        if (spatial)
+        {
+            _spatialRootObjectsByGuid[record.ServerGuid] = record;
+        }
+        else if (current
+                 || (_spatialRootObjectsByGuid.TryGetValue(record.ServerGuid, out var indexedRoot)
+                     && ReferenceEquals(indexedRoot, record)))
+        {
+            _spatialRootObjectsByGuid.Remove(record.ServerGuid);
+        }
+
         if (record.WorldEntity is { } entity)
         {
             if (spatial && record.AnimationRuntime is { } animation)
@@ -1455,6 +1957,12 @@ public sealed class LiveEntityRuntime
 
     private void RemoveSpatialRuntimeIndexes(LiveEntityRecord record)
     {
+        if (_spatialRootObjectsByGuid.TryGetValue(record.ServerGuid, out var indexedRoot)
+            && ReferenceEquals(indexedRoot, record))
+        {
+            _spatialRootObjectsByGuid.Remove(record.ServerGuid);
+        }
+
         if (record.WorldEntity is { } entity
             && _spatialAnimationsByLocalId.TryGetValue(entity.Id, out var indexedAnimation)
             && ReferenceEquals(indexedAnimation, record.AnimationRuntime))
@@ -1506,13 +2014,46 @@ public sealed class LiveEntityRuntime
             return;
 
         bool wasVisible = record.IsSpatiallyVisible;
+        bool wasOrdinaryRoot = _spatialRootObjectsByGuid.TryGetValue(
+                serverGuid,
+                out LiveEntityRecord? indexedRoot)
+            && ReferenceEquals(indexedRoot, record);
         record.IsSpatiallyVisible = visible;
+        bool isOrdinaryRoot = record.ProjectionKind is LiveEntityProjectionKind.World
+            && record.IsSpatiallyProjected
+            && visible
+            && record.FullCellId != 0;
+        if (_rebucketingGuid != serverGuid && wasOrdinaryRoot != isOrdinaryRoot)
+        {
+            if (isOrdinaryRoot)
+                record.ResetObjectClockForEnterWorld(
+                    (record.FinalPhysicsState & PhysicsStateFlags.Static) != 0);
+            else
+                record.SuspendObjectClock();
+            SynchronizePhysicsBodyActiveState(record);
+        }
         RefreshSpatialRuntimeIndexes(record);
         RefreshPresentation(record);
         if (_rebucketingGuid != serverGuid && wasVisible != visible)
             PublishProjectionVisibilityChanged(record, visible);
     }
 
+    private static void SynchronizePhysicsBodyActiveState(
+        LiveEntityRecord record)
+    {
+        if (record.PhysicsBody is not { } body)
+            return;
+        if (record.ObjectClock.IsActive)
+            body.TransientState |= TransientStateFlags.Active;
+        else
+            body.TransientState &= ~TransientStateFlags.Active;
+    }
+
+    private static bool IsFinite(Vector3 value) =>
+        float.IsFinite(value.X)
+        && float.IsFinite(value.Y)
+        && float.IsFinite(value.Z);
+
     private void PublishProjectionVisibilityChanged(LiveEntityRecord record, bool visible)
     {
         Delegate[] subscribers = ProjectionVisibilityChanged?.GetInvocationList()
@@ -1617,6 +2158,7 @@ public sealed class LiveEntityRuntime
         _spatialAnimationsByLocalId.Clear();
         _spatialRemoteMotionByGuid.Clear();
         _spatialProjectilesByGuid.Clear();
+        _spatialRootObjectsByGuid.Clear();
         ParentAttachments.Clear();
         _inbound.Clear();
         _spatial.ClearLiveEntityLifetimeState();

src/AcDream.App/World/LiveEntityRuntimeViews.cs

@@ -52,21 +52,25 @@ public sealed class LiveEntityAnimationRuntimeView<TAnimation>
             && runtime.ClearAnimationRuntime(guid);
     }
 
+    /// <summary>Copies only currently resident local IDs without boxing a
+    /// dictionary-key enumerator.</summary>
+    public void CopySpatialIdsTo(HashSet<uint> destination)
+    {
+        ArgumentNullException.ThrowIfNull(destination);
+        LiveEntityRuntime? runtime = _runtime();
+        if (runtime is null)
+            destination.Clear();
+        else
+            runtime.CopySpatialAnimationLocalIdsTo(destination);
+    }
+
     public Enumerator GetEnumerator()
     {
-        _iterationSnapshot.Clear();
         LiveEntityRuntime? runtime = _runtime();
-        if (runtime is not null)
-        {
-            foreach (var pair in runtime.SpatialAnimationRuntimes)
-            {
-                if (pair.Value is TAnimation typed)
-                {
-                    _iterationSnapshot.Add(
-                        new KeyValuePair<uint, TAnimation>(pair.Key, typed));
-                }
-            }
-        }
+        if (runtime is null)
+            _iterationSnapshot.Clear();
+        else
+            runtime.CopySpatialAnimationRuntimesTo(_iterationSnapshot);
 
         return new Enumerator(runtime, _iterationSnapshot);
     }

src/AcDream.App/World/RetailInboundEventDispatcher.cs

@@ -0,0 +1,123 @@
+namespace AcDream.App.World;
+
+/// <summary>
+/// Serializes retail network/event mutations on the update thread.
+/// </summary>
+/// <remarks>
+/// Retail SmartBox dispatch and CPhysics::UseTime never interleave two packet
+/// bodies on one CPhysicsObj call stack. App callbacks are synchronous, so an
+/// observer can otherwise re-enter a session event while an older packet or
+/// object quantum is still unwinding. Nested work is retained in arrival
+/// order and drained only after the current operation reaches its complete
+/// retail tail. This preserves the protocol's independent timestamp channels:
+/// an accepted State never cancels an accepted Position merely because their
+/// callbacks touched the same object.
+/// </remarks>
+internal sealed class RetailInboundEventDispatcher
+{
+    private interface IPendingOperation
+    {
+        void Invoke();
+    }
+
+    private sealed class PendingAction(Action operation) : IPendingOperation
+    {
+        public void Invoke() => operation();
+    }
+
+    private sealed class PendingStateOperation<TReceiver, TState>(
+        TReceiver receiver,
+        TState state,
+        Action<TReceiver, TState> operation) : IPendingOperation
+    {
+        public void Invoke() => operation(receiver, state);
+    }
+
+    private readonly Queue<IPendingOperation> _pending = new();
+    private bool _isDraining;
+
+    internal int PendingCount => _pending.Count;
+    internal bool IsDraining => _isDraining;
+
+    internal void Run(Action operation)
+    {
+        ArgumentNullException.ThrowIfNull(operation);
+        if (_isDraining)
+        {
+            _pending.Enqueue(new PendingAction(operation));
+            return;
+        }
+
+        _isDraining = true;
+        try
+        {
+            operation();
+            DrainPending();
+        }
+        catch
+        {
+            // A failed packet cannot leave later packets stranded for an
+            // unrelated future frame, or replay them after partial teardown.
+            _pending.Clear();
+            throw;
+        }
+        finally
+        {
+            _isDraining = false;
+        }
+    }
+
+    /// <summary>
+    /// Allocation-free non-reentrant dispatch for hot frame and packet paths.
+    /// A heterogeneous wrapper is created only when an operation genuinely
+    /// re-enters the active retail FIFO and therefore must outlive this call.
+    /// </summary>
+    internal void Run<TReceiver, TState>(
+        TReceiver receiver,
+        TState state,
+        Action<TReceiver, TState> operation)
+    {
+        ArgumentNullException.ThrowIfNull(operation);
+        if (_isDraining)
+        {
+            _pending.Enqueue(
+                new PendingStateOperation<TReceiver, TState>(
+                    receiver,
+                    state,
+                    operation));
+            return;
+        }
+
+        _isDraining = true;
+        try
+        {
+            operation(receiver, state);
+            DrainPending();
+        }
+        catch
+        {
+            _pending.Clear();
+            throw;
+        }
+        finally
+        {
+            _isDraining = false;
+        }
+    }
+
+    private void DrainPending()
+    {
+        while (_pending.TryDequeue(out IPendingOperation? next))
+            next.Invoke();
+    }
+
+    internal void Clear()
+    {
+        if (_isDraining)
+        {
+            throw new InvalidOperationException(
+                "Inbound event work cannot be cleared while its retail FIFO is active.");
+        }
+        _pending.Clear();
+    }
+}