 public extension Action {
 
     public func execute(input: Input) -> Observable<Element> {
-        // Send input to `self.inputs` to see all inputs in `self.inputs`.
-        skipsInputs.value = true
+        let buffer = ReplaySubject<Element>.createUnbounded()
+        let error = errors
+            .flatMap { error -> Observable<Element> in
+                if case .UnderlyingError(let error) = error {
+                    throw error
+                } else {
+                    return Observable.empty()
+                }
+            }
+        
+        Observable
+            .of(elements, error)
+            .merge()
+            .takeUntil(_completed)
+            .bindTo(buffer)
+            .addDisposableTo(disposeBag)
+        
         inputs.onNext(input)
-        skipsInputs.value = false
-
-        return _execute(input)
+        
+        return buffer.asObservable()
     }
 
     private func _execute(input: Input) -> Observable<Element> {

    /// Inputs that execute the action.
    /// Inputs via execute() also appear in this subject.
    public let inputs = PublishSubject<Input>()
    private let skipsInputs = Variable(false)

    /// Errors aggrevated from invocations of execute(). 
    /// Delivered on whatever scheduler they were sent from.

                onError: {[weak self] error in
                    self?._errors.onNext(ActionError.UnderlyingError(error))
                },
                onCompleted: {[weak self] in
                    self?._completed.onNext()
                },
                onDisposed: {[weak self] in
                    self?.doLocked { self?._executing.value = false }
                })
