diff --git a/PowerKit.Tests/ObservableTests.cs b/PowerKit.Tests/ObservableTests.cs
index 9437621..bfde1a3 100644
--- a/PowerKit.Tests/ObservableTests.cs
+++ b/PowerKit.Tests/ObservableTests.cs
@@ -6,19 +6,6 @@
 
 namespace PowerKit.Tests;
 
-file class FakeObserver<T>(
-    Action<T>? onNext = null,
-    Action<Exception>? onError = null,
-    Action? onCompleted = null
-) : IObserver<T>
-{
-    public void OnNext(T value) => onNext?.Invoke(value);
-
-    public void OnError(Exception error) => onError?.Invoke(error);
-
-    public void OnCompleted() => onCompleted?.Invoke();
-}
-
 public class ObservableTests
 {
     [Fact]
@@ -34,7 +21,7 @@ public void Observable_Create_Subscribe_Test()
 
         // Act
         subscribed.Should().BeFalse();
-        observable.Subscribe(new FakeObserver<int>());
+        observable.Subscribe(Observer.Create<int>());
 
         // Assert
         subscribed.Should().BeTrue();
@@ -55,7 +42,7 @@ public void Observable_Create_OnNext_Test()
         });
 
         // Act
-        observable.Subscribe(new FakeObserver<int>(received.Add));
+        observable.Subscribe(Observer.Create<int>(onNext: received.Add));
 
         // Assert
         received.Should().Equal(1, 2, 3);
@@ -73,11 +60,11 @@ public void Observable_Create_OnError_Test()
         });
 
         // Act
-        observable.Subscribe(new FakeObserver<int>(onError: ex => receivedError = ex));
+        observable.Subscribe(Observer.Create<int>(onError: ex => receivedError = ex));
 
         // Assert
         receivedError.Should().BeOfType<InvalidOperationException>();
-        receivedError.Message.Should().Be("test error");
+        receivedError!.Message.Should().Be("test error");
     }
 
     [Fact]
@@ -92,7 +79,7 @@ public void Observable_Create_OnCompleted_Test()
         });
 
         // Act
-        observable.Subscribe(new FakeObserver<int>(onCompleted: () => completed = true));
+        observable.Subscribe(Observer.Create<int>(onCompleted: () => completed = true));
 
         // Assert
         completed.Should().BeTrue();
@@ -103,15 +90,85 @@ public void Observable_Create_Dispose_Test()
     {
         // Arrange
         var disposed = false;
-        var observable = Observable.Create<int>(_ => Disposable.Create(() => disposed = true));
+        IObserver<int>? producer = null;
+        var received = new List<int>();
 
-        // Act
-        disposed.Should().BeFalse();
-        var subscription = observable.Subscribe(new FakeObserver<int>());
+        var observable = Observable.Create<int>(observer =>
+        {
+            producer = observer;
+            return Disposable.Create(() => disposed = true);
+        });
+
+        var subscription = observable.Subscribe(Observer.Create<int>(received.Add));
+
+        producer!.OnNext(1);
+        producer.OnNext(2);
+        producer.OnNext(3);
         subscription.Dispose();
+        producer.OnNext(4);
+        producer.OnNext(5);
 
         // Assert
         disposed.Should().BeTrue();
+        received.Should().Equal(1, 2, 3);
+    }
+
+    [Fact]
+    public void Observable_Create_Dispose_AfterOnCompleted_NoDoubleDispose_Test()
+    {
+        // Arrange
+        var disposeCount = 0;
+        IObserver<int>? producer = null;
+        var received = new List<int>();
+
+        var observable = Observable.Create<int>(observer =>
+        {
+            producer = observer;
+            return Disposable.Create(() => disposeCount++);
+        });
+
+        var subscription = observable.Subscribe(Observer.Create<int>(received.Add));
+
+        producer!.OnNext(1);
+        producer.OnNext(2);
+        producer.OnNext(3);
+        producer.OnCompleted();
+        subscription.Dispose();
+
+        // Assert
+        received.Should().Equal(1, 2, 3);
+        disposeCount.Should().Be(1);
+    }
+
+    [Fact]
+    public void Observable_Create_Dispose_NoEventAfterDispose_Test()
+    {
+        // Arrange
+        var completedCalled = false;
+        IObserver<int>? producer = null;
+        var received = new List<int>();
+
+        var observable = Observable.Create<int>(observer =>
+        {
+            producer = observer;
+            return Disposable.Null;
+        });
+
+        var subscription = observable.Subscribe(
+            Observer.Create<int>(onNext: received.Add, onCompleted: () => completedCalled = true)
+        );
+
+        producer!.OnNext(1);
+        producer.OnNext(2);
+        producer.OnNext(3);
+        subscription.Dispose();
+        producer.OnNext(4);
+        producer.OnNext(5);
+        producer.OnCompleted();
+
+        // Assert
+        received.Should().Equal(1, 2, 3);
+        completedCalled.Should().BeFalse();
     }
 
     [Fact]
@@ -129,7 +186,7 @@ public void Observable_CreateSynchronized_OnNext_Test()
         });
 
         // Act
-        observable.Subscribe(new FakeObserver<int>(received.Add));
+        observable.Subscribe(Observer.Create<int>(onNext: received.Add));
 
         // Assert
         received.Should().Equal(1, 2, 3);
@@ -168,9 +225,126 @@ public void Observable_CreateSynchronized_ThreadSafe_Test()
         });
 
         // Act
-        observable.Subscribe(new FakeObserver<int>(v => received.Add(v)));
+        observable.Subscribe(Observer.Create<int>(onNext: v => received.Add(v)));
 
         // Assert
         received.Should().HaveCount(threadCount * valuesPerThread);
     }
+
+    [Fact]
+    public void Observable_Create_AutoDetach_OnNext_Throw_DisposesSource_Test()
+    {
+        // Arrange: the subscribe callback emits 5 items synchronously; the observer
+        // collects them and throws on the third. Only 2 items are received and the
+        // source disposable must be disposed once subscribe returns.
+        var received = new List<int>();
+        var disposed = false;
+        var observable = Observable.Create<int>(observer =>
+        {
+            for (var i = 1; i <= 5; i++)
+            {
+                try
+                {
+                    observer.OnNext(i);
+                }
+                catch
+                {
+                    break;
+                }
+            }
+            return Disposable.Create(() => disposed = true);
+        });
+
+        // Act
+        observable.Subscribe(
+            Observer.Create<int>(onNext: v =>
+            {
+                if (v == 3)
+                    throw new InvalidOperationException();
+                received.Add(v);
+            })
+        );
+
+        // Assert
+        disposed.Should().BeTrue();
+        received.Should().Equal(1, 2);
+    }
+
+    [Fact]
+    public void Observable_Create_AutoDetach_OnError_Throw_DisposesSource_Test()
+    {
+        // Arrange
+        IObserver<int>? capturedObserver = null;
+        var disposed = false;
+        var observable = Observable.Create<int>(observer =>
+        {
+            capturedObserver = observer;
+            return Disposable.Create(() => disposed = true);
+        });
+        observable.Subscribe(
+            Observer.Create<int>(onError: _ => throw new InvalidOperationException("boom"))
+        );
+
+        // Act & Assert
+        var act = () => capturedObserver!.OnError(new Exception("source error"));
+        act.Should().Throw<InvalidOperationException>().WithMessage("boom");
+        disposed.Should().BeTrue();
+    }
+
+    [Fact]
+    public void Observable_Create_AutoDetach_OnCompleted_Throw_DisposesSource_Test()
+    {
+        // Arrange
+        IObserver<int>? capturedObserver = null;
+        var disposed = false;
+        var observable = Observable.Create<int>(observer =>
+        {
+            capturedObserver = observer;
+            return Disposable.Create(() => disposed = true);
+        });
+        observable.Subscribe(
+            Observer.Create<int>(onCompleted: () => throw new InvalidOperationException("boom"))
+        );
+
+        // Act & Assert
+        var act = () => capturedObserver!.OnCompleted();
+        act.Should().Throw<InvalidOperationException>().WithMessage("boom");
+        disposed.Should().BeTrue();
+    }
+
+    [Fact]
+    public void Observable_Create_AutoDetach_SuccessfulOnError_DisposesSource_Test()
+    {
+        // Arrange
+        var disposed = false;
+        var observable = Observable.Create<int>(observer =>
+        {
+            observer.OnError(new Exception("source error"));
+            return Disposable.Create(() => disposed = true);
+        });
+
+        // Act
+        observable.Subscribe(Observer.Create<int>(onError: _ => { }));
+
+        // Assert
+        disposed.Should().BeTrue();
+    }
+
+    [Fact]
+    public void Observable_Create_AutoDetach_SuccessfulOnCompleted_DisposesSource_Test()
+    {
+        // Arrange
+        var disposed = false;
+        var observable = Observable.Create<int>(observer =>
+        {
+            observer.OnCompleted();
+            return Disposable.Create(() => disposed = true);
+        });
+
+        // Act
+        observable.Subscribe(Observer.Create<int>());
+
+        // Assert
+        disposed.Should().BeTrue();
+    }
 }
diff --git a/PowerKit/AutoDetachObserver.cs b/PowerKit/AutoDetachObserver.cs
new file mode 100644
index 0000000..a8d3e42
--- /dev/null
+++ b/PowerKit/AutoDetachObserver.cs
@@ -0,0 +1,112 @@
+#if !NETFRAMEWORK || NET45_OR_GREATER
+using System;
+using System.Threading;
+
+namespace PowerKit;
+
+/// <summary>
+/// An observer that automatically disposes the upstream source subscription when a terminal
+/// event (<see cref="OnError" /> or <see cref="OnCompleted" />) is received, when any
+/// observer callback throws, or when the subscription is disposed externally.
+/// All observer methods are no-ops once the subscription has been stopped.
+/// </summary>
+internal class AutoDetachObserver<T>(IObserver<T> observer) : IObserver<T>, IDisposable
+{
+    private readonly Lock _lock = new();
+    private IDisposable? _disposable;
+    private bool _isUnsubscribedOrAbandoned;
+
+    internal void SetSubscription(IDisposable disposable)
+    {
+        IDisposable? toDispose;
+        lock (_lock)
+        {
+            if (_isUnsubscribedOrAbandoned)
+            {
+                toDispose = disposable;
+            }
+            else
+            {
+                _disposable = disposable;
+                toDispose = null;
+            }
+        }
+
+        toDispose?.Dispose();
+    }
+
+    private void DisposeSource()
+    {
+        IDisposable? disposable;
+        lock (_lock)
+        {
+            disposable = _disposable;
+            _disposable = null;
+            _isUnsubscribedOrAbandoned = true;
+        }
+
+        disposable?.Dispose();
+    }
+
+    /// <inheritdoc />
+    public void OnNext(T value)
+    {
+        lock (_lock)
+        {
+            if (_isUnsubscribedOrAbandoned)
+                return;
+        }
+
+        try
+        {
+            observer.OnNext(value);
+        }
+        catch
+        {
+            DisposeSource();
+            throw;
+        }
+    }
+
+    /// <inheritdoc />
+    public void OnError(Exception error)
+    {
+        lock (_lock)
+        {
+            if (_isUnsubscribedOrAbandoned)
+                return;
+        }
+
+        try
+        {
+            observer.OnError(error);
+        }
+        finally
+        {
+            DisposeSource();
+        }
+    }
+
+    /// <inheritdoc />
+    public void OnCompleted()
+    {
+        lock (_lock)
+        {
+            if (_isUnsubscribedOrAbandoned)
+                return;
+        }
+
+        try
+        {
+            observer.OnCompleted();
+        }
+        finally
+        {
+            DisposeSource();
+        }
+    }
+
+    /// <inheritdoc />
+    public void Dispose() => DisposeSource();
+}
+#endif
diff --git a/PowerKit/Observable.cs b/PowerKit/Observable.cs
index 441152e..481ca76 100644
--- a/PowerKit/Observable.cs
+++ b/PowerKit/Observable.cs
@@ -17,7 +17,13 @@ public static class Observable
     /// Creates an observable that invokes the specified subscribe function when subscribed to.
     /// </summary>
     public static IObservable<T> Create<T>(Func<IObserver<T>, IDisposable> subscribe) =>
-        new Observable<T>(subscribe);
+        new Observable<T>(observer =>
+        {
+            var autoDetach = new AutoDetachObserver<T>(observer);
+            var disposable = subscribe(autoDetach);
+            autoDetach.SetSubscription(disposable);
+            return autoDetach;
+        });
 
     /// <summary>
     /// Creates an observable that invokes the specified subscribe function when subscribed to,
diff --git a/PowerKit/Observer.cs b/PowerKit/Observer.cs
new file mode 100644
index 0000000..de4e790
--- /dev/null
+++ b/PowerKit/Observer.cs
@@ -0,0 +1,33 @@
+#if !NETFRAMEWORK || NET45_OR_GREATER
+using System;
+
+namespace PowerKit;
+
+file class DelegateObserver<T>(Action<T>? onNext, Action<Exception>? onError, Action? onCompleted)
+    : IObserver<T>
+{
+    public void OnNext(T value) => onNext?.Invoke(value);
+
+    public void OnError(Exception error) => onError?.Invoke(error);
+
+    public void OnCompleted() => onCompleted?.Invoke();
+}
+
+/// <summary>
+/// Provides utility methods for creating <see cref="IObserver{T}" /> instances.
+/// </summary>
+public static class Observer
+{
+    /// <summary>
+    /// Creates an observer from the specified delegate callbacks.
+    /// </summary>
+    /// <remarks>
+    /// Any callback left as <see langword="null" /> is treated as a no-op.
+    /// </remarks>
+    public static IObserver<T> Create<T>(
+        Action<T>? onNext = null,
+        Action<Exception>? onError = null,
+        Action? onCompleted = null
+    ) => new DelegateObserver<T>(onNext, onError, onCompleted);
+}
+#endif