Migrate from lock-based synchronization to Observable/Rx patterns

[juliomenendez]

Summary

This PR migrates the Agents-for-net codebase from lock-based synchronization to Observable/Rx patterns (System.Reactive 6.0.0). The migration is designed as a series of atomic commits that maintain project stability.

Changes by Component

• System.Reactive Infrastructure: Add System.Reactive 6.0.0 package and ObservableBridge utility for IAsyncEnumerable/IObservable interop
• RouteList.cs: Replace obsolete ReaderWriterLock with BehaviorSubject<ImmutableList<T>> for lock-free route management
• MemoryStorage.cs: Replace lock(_syncroot) with ConcurrentDictionary and fix _eTag++ race condition using Interlocked.Increment
• MemoryTranscriptStore.cs: Replace nested Dictionary with ConcurrentDictionary<string, ConcurrentDictionary<string, ImmutableList<IActivity>>>
• BackgroundTaskQueue.cs: Convert to Observable pattern using Subject<Unit> for signaling while maintaining queue semantics
• ActivityTaskQueue.cs: Fix thread-safety bug (bool _stopped race condition) and convert to full Observable pattern with drain completion tracking
• HostedTaskService + HostedActivityService: Replace ReaderWriterLockSlim with BehaviorSubject<bool> for graceful shutdown
• StreamingResponse.cs: Complete rewrite replacing 7 lock(this) statements with BehaviorSubject<StreamState>, Timer with Observable.Interval, and proper Rx pipeline

Issues Fixed

• ActivityTaskQueue.cs:24 - _stopped boolean race condition (now uses Interlocked)
• RouteList.cs:12 - Obsolete ReaderWriterLock (now uses BehaviorSubject)
• StreamingResponse.cs - Timer callbacks within lock blocks (potential deadlock)
• MemoryStorage.cs:152 - ETag increment not thread-safe (now uses Interlocked)

Key Benefits

• Eliminates lock contention with lock-free data structures
• Clear state management via Rx BehaviorSubject
• Built-in backpressure handling via Rx operators
• Proper cancellation support via TakeUntil
• No more lock(this) anti-pattern

Test plan

• All existing unit tests pass (477+ tests)
• Added concurrency tests for ActivityTaskQueue.Stop()
• Verified builds for both net8.0 and netstandard2.0

🤖 Generated with Claude Code

[Copilot]

Pull request overview

This PR migrates the Agents-for-net codebase from traditional lock-based synchronization to Observable/Rx patterns using System.Reactive 6.0.0. The migration aims to eliminate lock contention and improve concurrency handling across multiple components including storage layers, background task queues, and streaming response handling.

Changes:

• Introduced System.Reactive 6.0.0 and ObservableBridge utility for Observable/IAsyncEnumerable interop
• Replaced lock-based synchronization with ConcurrentDictionary, BehaviorSubject, and Interlocked operations
• Migrated BackgroundTaskQueue and ActivityTaskQueue to use Subject-based signaling
• Completely rewrote StreamingResponse with Rx pipeline replacing Timer and lock(this) patterns
• Added comprehensive concurrency tests for ActivityTaskQueue

Reviewed changes

Copilot reviewed 19 out of 19 changed files in this pull request and generated 21 comments.

Show a summary per file

File	Description
Directory.Packages.props	Added System.Reactive 6.0.0 and System.Collections.Immutable 10.0.2 package versions
ObservableBridge.cs	New utility for IAsyncEnumerable/IObservable conversion (has circular call bug)
MemoryStorage.cs	Replaced lock with ConcurrentDictionary and Interlocked.Increment for thread-safe ETag generation
MemoryTranscriptStore.cs	Migrated to ConcurrentDictionary with ImmutableList for lock-free transcript storage
BackgroundTaskQueue.cs	Converted to Subject-based signaling (has race condition with multiple consumers)
ActivityTaskQueue.cs	Fixed boolean race condition with Interlocked, converted to Observable pattern (has multiple issues)
HostedTaskService.cs	Replaced ReaderWriterLockSlim with BehaviorSubject for shutdown coordination
HostedActivityService.cs	Replaced ReaderWriterLockSlim with BehaviorSubject for shutdown coordination
RouteList.cs	Replaced obsolete ReaderWriterLock with BehaviorSubject and ImmutableList (has compare-and-swap bug)
StreamingResponse.cs	Complete rewrite with BehaviorSubject for state and Observable.Interval (has multiple race conditions)
ActivityTaskQueueTests.cs	Added concurrency tests for Stop() method and thread safety
Multiple test projects	Added System.Reactive package reference

Comments suppressed due to low confidence (1)

src/libraries/Storage/Microsoft.Agents.Storage.Transcript/MemoryTranscriptStore.cs:145

• These 'if' statements can be combined.

            if (_channels.TryGetValue(channelId, out var channel))
            {
                if (channel.TryGetValue(conversationId, out var transcript))
                {
                    if (continuationToken != null)
                    {
                        pagedResult.Items = transcript
                            .OrderBy(a => a.Timestamp)
                            .Where(a => a.Timestamp >= startDate)
                            .SkipWhile(a => a.Id != continuationToken)
                            .Skip(1)
                            .Take(20)
                            .ToArray();

                        if (pagedResult.Items.Count == 20)
                        {
                            pagedResult.ContinuationToken = pagedResult.Items.Last().Id;
                        }
                    }
                    else
                    {
                        pagedResult.Items = transcript
                            .OrderBy(a => a.Timestamp)
                            .Where(a => a.Timestamp >= startDate)
                            .Take(20)
                            .ToArray();

                        if (pagedResult.Items.Count == 20)
                        {
                            pagedResult.ContinuationToken = pagedResult.Items.Last().Id;
                        }
                    }
                }
            }

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[juliomenendez]

@tracyboehrer this is the replacement for #542

[tracyboehrer]

Can we limit this to StreamingResponse? For scope reduction, and we will likely be changing the background queue up separately.