     /// Inputs that execute the action.
     /// Inputs via execute() also appear in this subject.
     public let inputs = PublishSubject<Input>()
-    private let skipsInputs = Variable(false)
+    private let _completed = PublishSubject<Void>()
 
     /// 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: nil,
                onDisposed: {[weak self] in
                    self?.doLocked { self?._executing.value = false }
                })

public extension Action {

    public func execute(input: Input) -> Observable<Element> {
        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)
        
        return buffer.asObservable()
    }

    private func _execute(input: Input) -> Observable<Element> {
