Fix SSE integration test comparison logic to use correct series pattern

.devcontainer/devcontainer.json

@@ -21,6 +21,7 @@
     "vscode": {
       "settings": {
         "chat.tools.global.autoApprove": true,
+        "dotnet.defaultSolution": "Stock.Indicators.sln",
         "terminal.integrated.defaultProfile.linux": "zsh"
       },
       "extensions": [

docs/features/stream.md

@@ -106,7 +106,7 @@ quoteHub.Add(quote1);
 quoteHub.Add(quote2);
 
 // late-arriving data with earlier timestamp
-quoteHub.Insert(lateQuote);  // triggers recalculation
+quoteHub.Add(lateQuote);  // triggers recalculation in dependent hubs
 
 // remove incorrect quote
 quoteHub.Remove(badQuote);   // triggers recalculation
@@ -125,8 +125,8 @@ The hub automatically handles state rollback and recalculation when data arrives
 
 Stream hubs use internal locking to protect cache operations during rebuild and rollback scenarios:
 
-- **Internal cache operations** are thread-safe (Insert, RemoveAt, RemoveRange, Rebuild)
-- **External access requires synchronization** when multiple threads call Add, Insert, or Remove
+- **Internal cache operations** are thread-safe (Add, RemoveAt, RemoveRange, Rebuild)
+- **External access requires synchronization** when multiple threads call Add or Remove
 - **Single-threaded usage** requires no additional synchronization
 - **Multi-threaded usage** should synchronize external calls to hub methods
 

docs/guide.md

@@ -451,7 +451,7 @@ quoteHub.Add(quote1);
 quoteHub.Add(quote2);
 
 // Late-arriving data with earlier timestamp
-quoteHub.Insert(lateQuote);  // Triggers recalculation
+quoteHub.Add(lateQuote);  // Triggers recalculation in dependent hubs
 
 // Remove incorrect quote
 quoteHub.Remove(badQuote);   // Triggers recalculation
@@ -747,7 +747,7 @@ await Task.WhenAll(
 **You need locks or channels when:**
 
 - Sharing a single hub instance across multiple threads
-- Multiple threads call `.Add()`, `.Insert()`, or `.Remove()` on the same hub
+- Do not call `.Add()` or `.Remove()` on the same hub from multiple threads
 - Reading results from one thread while another thread updates the hub
 - Processing quotes from multiple concurrent sources into one hub
 

docs/migration.md

@@ -347,6 +347,7 @@ Popular indicators with complete streaming documentation:
 | `UseResult` | `TimeValue` | Type renamed |
 | `SmaAnalysis` | `SmaAnalysisResult` | Type renamed |
 | `UlcerIndexResult.UI` | `UlcerIndexResult.UlcerIndex` | Property renamed |
+| `StreamHub.Insert()` | `StreamHub.Add()` | Insert removed; use Add |
 | `SyncSeries()` | (removed) | Use manual alignment |
 | `Find()` / `FindIndex()` | LINQ methods | Use `.FirstOrDefault()` etc. |
 | `GetBaseQuote()` | `Use(CandlePart)` | Use utility instead |

docs/performance.md

@@ -170,7 +170,7 @@ StreamHub style adds observable patterns and state management:
 Series (502 quotes):      ~25μs total
 StreamHub (502 quotes):   ~32μs total (~28% overhead)
 Per-quote latency:        ~64ns average
-Rollback (Insert):        ~2-5μs for state rebuild
+Rollback (late Add):      ~2-5μs for state rebuild
 ```
 
 **Scaling Characteristics:**
@@ -206,7 +206,7 @@ Real-time performance targets for trading applications:
 | Single indicator per quote         | <100μs  | 60-80μs             |
 | 5 indicators on hub per quote      | <500μs  | 300-400μs           |
 | Complex chains (EMA→RSI→Slope)     | <200μs  | 120-150μs           |
-| State rebuild (Insert/Remove)      | <5ms    | 2-3ms               |
+| State rebuild (Add/Remove)         | <5ms    | 2-3ms               |
 
 ### When to Choose Each Style
 

docs/plans/test-pruning.md

@@ -123,7 +123,7 @@ For the remaining 75 indicators, add validation following these patterns:
 - **T3**: `ValidateCacheSize(lookbackPeriods * 6, Name);`
 - **Multi-parameter**: Calculate `requiredWarmup` based on indicator-specific logic, then call `ValidateCacheSize(requiredWarmup, Name);`
 
-Insert validation call immediately before `Reinitialize()` in each indicator's constructor.
+Add validation call immediately before `Reinitialize()` in each indicator's constructor.
 
 ---
 Last updated: January 26, 2026

src/_common/Enums/Act.cs

@@ -15,13 +15,8 @@ internal enum Act
     /// </summary>
     Ignore = 1,
 
-    /// <summary>
-    /// Insert item without rebuilding cache.
-    /// </summary>
-    Insert = 2,
-
     /// <summary>
     /// Reset and rebuild from marker position.
     /// </summary>
-    Rebuild = 3
+    Rebuild = 2
 }

src/_common/QuotePart/QuotePart.StreamHub.cs

@@ -19,6 +19,24 @@ CandlePart candlePart
     /// <inheritdoc/>
     public CandlePart CandlePartSelection { get; init; }
 
+    /// <inheritdoc/>
+    public override void OnAdd(IQuote item, bool notify, int? indexHint)
+    {
+        // Lock to prevent concurrent cache access.
+        lock (CacheLock)
+        {
+            (TimeValue result, int index) = ToIndicator(item, indexHint);
+
+            if (index >= 0 && index < Cache.Count)
+            {
+                InsertWithoutRebuild(result, index, notify);
+                return;
+            }
+
+            AppendCache(result, notify);
+        }
+    }
+
 
     /// <inheritdoc/>
     protected override (TimeValue result, int index)

src/_common/Quotes/Quote.StreamHub.cs

@@ -84,6 +84,17 @@ public override void OnAdd(IQuote item, bool notify, int? indexHint)
     {
         ArgumentNullException.ThrowIfNull(item);
 
+        // Reject additions that precede the current cache timeline
+        // (applies to both standalone and non-standalone QuoteHub)
+        lock (CacheLock)
+        {
+            if (Cache.Count > 0 && item.Timestamp < Cache[0].Timestamp)
+            {
+                // Silently ignore - this prevents indeterminate gaps in the timeline
+                return;
+            }
+        }
+
         // for non-standalone QuoteHub, use standard behavior (which handles locking)
         if (!_isStandalone)
         {
@@ -94,9 +105,19 @@ public override void OnAdd(IQuote item, bool notify, int? indexHint)
         // Lock for standalone QuoteHub operations
         lock (CacheLock)
         {
+
             // get result and position
             (IQuote result, int index) = ToIndicator(item, indexHint);
 
+            // Reject modifications that would affect indices before MinCacheSize
+            // to prevent corrupted rebuilds in subscribers
+            // This includes both insertions and same-timestamp replacements
+            if (index >= 0 && index < MinCacheSize && index < Cache.Count)
+            {
+                // Silently ignore all modifications before MinCacheSize
+                return;
+            }
+
             // check if this is a same-timestamp update (not a new item at the end)
             if (Cache.Count > 0 && index < Cache.Count && Cache[index].Timestamp == result.Timestamp)
             {
@@ -119,6 +140,23 @@ public override void OnAdd(IQuote item, bool notify, int? indexHint)
             }
             else
             {
+                // if out-of-order insert, insert and trigger rebuild
+                if (index >= 0 && index < Cache.Count)
+                {
+                    Cache.Insert(index, result);
+
+                    // For standalone QuoteHub, notify observers to rebuild from this timestamp
+                    // For non-standalone, this won't be reached due to earlier branch
+                    if (notify)
+                    {
+                        // Notify observers directly - they will rebuild from the updated cache
+                        // No need to call Rebuild on QuoteHub itself since cache is already updated
+                        NotifyObserversOnRebuild(result.Timestamp);
+                    }
+
+                    return;
+                }
+
                 // standard add behavior for new items
                 AppendCache(result, notify);
             }

src/_common/StreamHub/IStreamHub.cs

@@ -60,19 +60,6 @@ public interface IStreamHub<in TIn, out TOut> : IStreamObserver<TIn>, IStreamObs
     /// </param>
     void Add(IEnumerable<TIn> batchIn);
 
-    /// <summary>
-    /// Insert a new item without rebuilding the cache.
-    /// </summary>
-    /// <remarks>
-    /// This is used in situations when inserting an older item
-    /// and where newer cache entries do not need to be rebuilt.
-    /// Typically, this is only used for provider-only hubs.
-    /// </remarks>
-    /// <param name="newIn">
-    /// Item to insert
-    /// </param>
-    void Insert(TIn newIn);
-
     /// <summary>
     /// Delete an item from the cache, from a specific position.
     /// </summary>

src/_common/StreamHub/IStreamObservable.cs

@@ -52,6 +52,14 @@ public interface IStreamObservable<out T>
     /// </summary>
     int MaxCacheSize { get; }
 
+    /// <summary>
+    /// Gets the minimum cache size required by this hub or its subscribers.
+    /// This value represents the maximum minimum cache size requirement from all subscribers.
+    /// Additions that would modify index values before this position are rejected to prevent
+    /// corrupted rebuilds.
+    /// </summary>
+    int MinCacheSize { get; }
+
     /// <summary>
     /// Gets the current number of subscribers.
     /// </summary>

src/_common/StreamHub/Providers/BaseProvider.cs

@@ -43,6 +43,9 @@ public class BaseProvider<T>(int maxCacheSize = 0)
     /// <inheritdoc/>
     public int MaxCacheSize { get; } = maxCacheSize;
 
+    /// <inheritdoc/>
+    public int MinCacheSize => 0; // Base provider has no minimum cache requirement
+
     /// <inheritdoc />
     public int ObserverCount => 0;
 

src/_common/StreamHub/StreamHub.Observable.cs

@@ -6,6 +6,13 @@
 {
     private readonly HashSet<IStreamObserver<TOut>> _observers = [];
 
+    /// <summary>
+    /// Stores the hub's own minimum cache size requirement (baseline).
+    /// This value represents the warmup periods needed by this hub itself,
+    /// independent of its subscribers.
+    /// </summary>
+    private int _minCacheSizeBaseline;
+
     // PROPERTIES
 
     /// <inheritdoc/>
@@ -14,6 +21,9 @@
     /// <inheritdoc/>
     public int MaxCacheSize { get; private set; }
 
+    /// <inheritdoc/>
+    public int MinCacheSize { get; private set; }
+
     /// <inheritdoc/>
     public int ObserverCount => _observers.Count;
 
@@ -30,12 +40,46 @@
     public IDisposable Subscribe(IStreamObserver<TOut> observer)
     {
         _observers.Add(observer);
-        return new Unsubscriber(_observers, observer);
+
+        // Update MinCacheSize to the maximum of all subscribers
+        UpdateMinCacheSize();
+
+        return new Unsubscriber(_observers, observer, this);
     }
 
     /// <inheritdoc/>
     public bool Unsubscribe(IStreamObserver<TOut> observer)
-        => _observers.Remove(observer);
+    {
+        bool removed = _observers.Remove(observer);
+
+        // Re-evaluate MinCacheSize after unsubscribing
+        if (removed)
+        {
+            UpdateMinCacheSize();
+        }
+
+        return removed;
+    }
+
+    /// <summary>
+    /// Updates the MinCacheSize to the maximum of this hub's baseline requirement
+    /// and all subscribers' MinCacheSize values.
+    /// </summary>
+    private void UpdateMinCacheSize()
+    {
+        // Start from the hub's own baseline requirement
+        int maxMinCacheSize = _minCacheSizeBaseline;
+
+        foreach (IStreamObserver<TOut> observer in _observers)
+        {
+            if (observer is IStreamObservable<ISeries> observable)
+            {
+                maxMinCacheSize = Math.Max(maxMinCacheSize, observable.MinCacheSize);
+            }
+        }
+
+        MinCacheSize = maxMinCacheSize;
+    }
 
     /// <inheritdoc/>
     public void EndTransmission()
@@ -51,6 +95,9 @@
         }
 
         _observers.Clear();
+
+        // Reset to baseline when all subscribers are removed
+        MinCacheSize = _minCacheSizeBaseline;
     }
 
     /// <summary>
@@ -63,17 +110,28 @@
     /// <param name="observer">
     /// Your unique subscription as provided.
     /// </param>
+    /// <param name="hub">
+    /// The parent hub that needs MinCacheSize re-evaluation on unsubscribe.
+    /// </param>
     private sealed class Unsubscriber(
         ISet<IStreamObserver<TOut>> observers,
-        IStreamObserver<TOut> observer) : IDisposable
+        IStreamObserver<TOut> observer,
+        StreamHub<TIn, TOut> hub) : IDisposable
     {
         private readonly ISet<IStreamObserver<TOut>> _observers = observers;
         private readonly IStreamObserver<TOut> _observer = observer;
+        private readonly StreamHub<TIn, TOut> _hub = hub;
 
         /// <summary>
-        /// Remove single observer.
+        /// Remove single observer and update parent MinCacheSize.
         /// </summary>
-        public void Dispose() => _observers.Remove(_observer);
+        public void Dispose()
+        {
+            if (_observers.Remove(_observer))
+            {
+                _hub.UpdateMinCacheSize();
+            }
+        }
     }
 
     /// <summary>