diff --git a/dvc/exceptions.py b/dvc/exceptions.py
index b26481fc6..565ce4023 100644
--- a/dvc/exceptions.py
+++ b/dvc/exceptions.py
@@ -117,3 +117,17 @@ class DvcParserError(DvcException):
     """Base class for CLI parser errors."""
     def __init__(self):
         super(DvcParserError, self).__init__("parser error")
+
+
+class CyclicGraphError(DvcException):
+    def __init__(self, stages):
+        assert isinstance(stages, list)
+        stages = '\n'.join('\t- {}'.format(stage) for stage in stages)
+        msg = (
+            "you've introduced a cycle in your pipeline that involves"
+            ' the following stages:'
+            '\n'
+            '{stages}'
+            .format(stages=stages)
+        )
+        super(CyclicGraphError, self).__init__(msg)
diff --git a/dvc/project.py b/dvc/project.py
index ed1affbe4..74403171a 100644
--- a/dvc/project.py
+++ b/dvc/project.py
@@ -176,25 +176,15 @@ class Project(object):
     def install(self):
         self.scm.install()
 
-    def _check_cwd_specified_as_output(self, cwd, stages):
-        from dvc.exceptions import WorkingDirectoryAsOutputError
+    @staticmethod
+    def _check_cyclic_graph(graph):
+        import networkx as nx
+        from dvc.exceptions import CyclicGraphError
 
-        cwd_path = os.path.abspath(os.path.normpath(cwd))
+        cycles = list(nx.simple_cycles(graph))
 
-        for stage in stages:
-            for output in stage.outs:
-                if os.path.isdir(output.path) and output.path == cwd_path:
-                    raise WorkingDirectoryAsOutputError(cwd, stage.relpath)
-
-    def _check_output_duplication(self, outs, stages):
-        from dvc.exceptions import OutputDuplicationError
-
-        for stage in stages:
-            for o in stage.outs:
-                for out in outs:
-                    if o.path == out.path and o.stage.path != out.stage.path:
-                        stages = [o.stage.relpath, out.stage.relpath]
-                        raise OutputDuplicationError(o.path, stages)
+        if cycles:
+            raise CyclicGraphError(cycles[0])
 
     def add(self, fname, recursive=False):
         from dvc.stage import Stage
@@ -215,7 +205,6 @@ class Project(object):
         else:
             fnames = [fname]
 
-        all_stages = self.stages()
         stages = []
         self._files_to_git_add = []
         with self.state:
@@ -228,16 +217,22 @@ class Project(object):
                     stages.append(stage)
                     continue
 
-                self._check_output_duplication(stage.outs, all_stages)
-
                 stage.save()
-                stage.dump()
                 stages.append(stage)
 
+        self._check_dag(self.stages() + stages)
+
+        for stage in stages:
+            stage.dump()
+
         self._remind_to_git_add()
 
         return stages
 
+    def _check_dag(self, stages):
+        """Generate graph including the new stage to check for errors"""
+        self.graph(stages=stages)
+
     def remove(self, target, outs_only=False):
         from dvc.stage import Stage
 
@@ -389,10 +384,7 @@ class Project(object):
         if stage is None:
             return None
 
-        all_stages = self.stages()
-
-        self._check_cwd_specified_as_output(cwd, all_stages)
-        self._check_output_duplication(stage.outs, all_stages)
+        self._check_dag(self.stages() + [stage])
 
         self._files_to_git_add = []
         with self.state:
@@ -416,7 +408,7 @@ class Project(object):
         if stage is None:
             return None
 
-        self._check_output_duplication(stage.outs, self.stages())
+        self._check_dag(self.stages() + [stage])
 
         self._files_to_git_add = []
         with self.state:
@@ -1174,24 +1166,35 @@ class Project(object):
     def metrics_remove(self, path):
         self._metrics_modify(path, delete=True)
 
-    def graph(self, from_directory=None):
+    def graph(self, stages=None, from_directory=None):
         import networkx as nx
-        from dvc.exceptions import OutputDuplicationError
+        from dvc.exceptions import (OutputDuplicationError,
+                                    WorkingDirectoryAsOutputError)
 
         G = nx.DiGraph()
         G_active = nx.DiGraph()
-        stages = self.stages(from_directory)
-
+        stages = stages or self.stages(from_directory)
+        stages = [stage for stage in stages if stage]
         outs = []
-        outs_by_path = {}
+
+        for stage in stages:
+            for out in stage.outs:
+                existing = [o.stage for o in outs if o.path == out.path]
+
+                if existing:
+                    stages = [stage.relpath, existing[0].relpath]
+                    raise OutputDuplicationError(out.path, stages)
+
+                outs.append(out)
+
         for stage in stages:
-            for o in stage.outs:
-                existing = outs_by_path.get(o.path, None)
-                if existing is not None:
-                    stages = [o.stage.relpath, existing.stage.relpath]
-                    raise OutputDuplicationError(o.path, stages)
-                outs.append(o)
-                outs_by_path[o.path] = o
+            for out in outs:
+                overlaps = (stage.cwd == out.path
+                            or stage.cwd.startswith(out.path + os.sep))
+
+                if overlaps:
+                    raise WorkingDirectoryAsOutputError(stage.cwd,
+                                                        stage.relpath)
 
         # collect the whole DAG
         for stage in stages:
@@ -1215,12 +1218,14 @@ class Project(object):
                         G_active.add_node(dep_node, stage=dep_stage)
                         G_active.add_edge(node, dep_node)
 
+        self._check_cyclic_graph(G)
+
         return G, G_active
 
     def pipelines(self, from_directory=None):
         import networkx as nx
 
-        G, G_active = self.graph(from_directory)
+        G, G_active = self.graph(from_directory=from_directory)
 
         return [
             G.subgraph(c).copy()
diff --git a/dvc/stage.py b/dvc/stage.py
index 7bb718f7e..3512a1a9c 100644
--- a/dvc/stage.py
+++ b/dvc/stage.py
@@ -391,6 +391,8 @@ class Stage(object):
             if not overwrite and not prompt.confirm(msg):
                 raise StageFileAlreadyExistsError(stage.relpath)
 
+            os.unlink(path)
+
         return stage
 
     @staticmethod

graph: check for circular dependencies when parsing DAG
E.g. same way we check output duplication in project.py:graph(). We should be able to just search for cycles in the networkx graph and print the trace.
diff --git a/tests/test_repro.py b/tests/test_repro.py
index a943657e2..998b7eb96 100644
--- a/tests/test_repro.py
+++ b/tests/test_repro.py
@@ -17,6 +17,7 @@ from dvc.utils.compat import urljoin
 from dvc.remote.local import RemoteLOCAL
 from dvc.stage import Stage, StageFileDoesNotExistError
 from dvc.system import System
+from dvc.exceptions import CyclicGraphError, WorkingDirectoryAsOutputError
 
 from tests.basic_env import TestDvc
 from tests.test_data_cloud import _should_test_aws, TEST_AWS_REPO_BUCKET
@@ -53,6 +54,145 @@ class TestReproFail(TestRepro):
         self.assertNotEqual(ret, 0)
 
 
+class TestReproCyclicGraph(TestDvc):
+    def test(self):
+        self.dvc.run(deps=[self.FOO],
+                     outs=['bar.txt'],
+                     cmd='echo bar > bar.txt')
+
+        self.dvc.run(deps=['bar.txt'],
+                     outs=['baz.txt'],
+                     cmd='echo baz > baz.txt')
+
+        with open('cycle.dvc', 'w') as fd:
+            stage_dump = {
+                'cmd': 'echo baz > foo',
+                'deps': [{'path': 'baz.txt'}],
+                'outs': [{'path': self.FOO}],
+            }
+
+            yaml.safe_dump(stage_dump, fd, default_flow_style=False)
+
+        with self.assertRaises(CyclicGraphError):
+            self.dvc.reproduce('cycle.dvc')
+
+
+class TestReproWorkingDirectoryAsOutput(TestDvc):
+    """
+    |  stage.cwd  |  out.path | cwd as output |
+    |:-----------:|:---------:|:-------------:|
+    |     dir     |    dir    |      True     |
+    | dir/subdir/ |    dir    |      True     |
+    |     dir     |   dir-1   |     False     |
+    |      .      | something |     False     |
+    """
+    def test(self):
+        # File structure:
+        #       .
+        #       |-- dir1
+        #       |  |__ dir2.dvc         (out.path == ../dir2)
+        #       |__ dir2
+        #           |__ something.dvc    (stage.cwd == ./dir2)
+
+        os.mkdir(os.path.join(self.dvc.root_dir, 'dir1'))
+
+        self.dvc.run(
+            cwd='dir1',
+            outs=['../dir2'],
+            cmd='mkdir {path}'.format(path=os.path.join('..', 'dir2'))
+        )
+
+        faulty_stage_path = os.path.join('dir2', 'something.dvc')
+
+        with open(faulty_stage_path, 'w') as fd:
+            stage_dump = {
+                'cmd': 'echo something > something',
+                'outs': [{'path': 'something'}],
+            }
+
+            yaml.safe_dump(stage_dump, fd, default_flow_style=False)
+
+        with self.assertRaises(WorkingDirectoryAsOutputError):
+            self.dvc.reproduce(faulty_stage_path)
+
+    def test_nested(self):
+        from dvc.stage import Stage
+        #
+        #       .
+        #       |-- a
+        #       |  |__ nested
+        #       |     |__ dir
+        #       |       |__ error.dvc     (stage.cwd == 'a/nested/dir')
+        #       |__ b
+        #          |__ nested.dvc         (stage.out == 'a/nested')
+        dir1 = 'b'
+        dir2 = 'a'
+
+        os.mkdir(dir1)
+        os.mkdir(dir2)
+
+        nested_dir = os.path.join(dir2, 'nested')
+        out_dir = os.path.relpath(nested_dir, dir1)
+
+        nested_stage = self.dvc.run(
+            cwd=dir1,          # b
+            outs=[out_dir],    # ../a/nested
+            cmd='mkdir {path}'.format(path=out_dir)
+        )
+
+        os.mkdir(os.path.join(nested_dir, 'dir'))
+
+        error_stage_path = os.path.join(nested_dir, 'dir', 'error.dvc')
+
+        with open(error_stage_path, 'w') as fd:
+            stage_dump = {
+                'cmd': 'echo something > something',
+                'outs': [{'path': 'something'}],
+            }
+
+            yaml.safe_dump(stage_dump, fd, default_flow_style=False)
+
+        # NOTE: os.walk() walks in a sorted order and we need dir2 subdirs to
+        # be processed before dir1 to load error.dvc first.
+        with patch.object(Project, 'stages') as mock_stages:
+            mock_stages.return_value = [
+                nested_stage,
+                Stage.load(self.dvc, error_stage_path)
+            ]
+
+            with self.assertRaises(WorkingDirectoryAsOutputError):
+                self.dvc.reproduce(error_stage_path)
+
+    def test_similar_paths(self):
+        # File structure:
+        #
+        #       .
+        #       |-- something.dvc   (out.path == something)
+        #       |-- something
+        #       |__ something-1
+        #          |-- a
+        #          |__ a.dvc        (stage.cwd == something-1)
+
+        self.dvc.run(outs=['something'], cmd='mkdir something')
+
+        os.mkdir('something-1')
+
+        stage = os.path.join('something-1', 'a.dvc')
+
+        with open(stage, 'w') as fd:
+            stage_dump = {
+                'cmd': 'echo a > a',
+                'outs': [{'path': 'a'}],
+            }
+
+            yaml.safe_dump(stage_dump, fd, default_flow_style=False)
+
+        try:
+            self.dvc.reproduce(stage)
+        except WorkingDirectoryAsOutputError:
+            self.fail('should not raise WorkingDirectoryAsOutputError')
+
+
 class TestReproDepUnderDir(TestDvc):
     def test(self):
         stages = self.dvc.add(self.DATA_DIR)
diff --git a/tests/test_run.py b/tests/test_run.py
index 71f64ce45..3ccbb0f0f 100644
--- a/tests/test_run.py
+++ b/tests/test_run.py
@@ -12,6 +12,7 @@ from dvc.stage import StageFileBadNameError, MissingDep
 from dvc.stage import StageBadCwdError, StageFileAlreadyExistsError
 from dvc.exceptions import (OutputDuplicationError,
                             CircularDependencyError,
+                            CyclicGraphError,
                             ArgumentDuplicationError,
                             WorkingDirectoryAsOutputError)
 
@@ -122,6 +123,20 @@ class TestRunCircularDependency(TestDvc):
                          outs=['foo'],
                          fname='circular-dependency.dvc')
 
+    def test_graph(self):
+        self.dvc.run(deps=[self.FOO],
+                     outs=['bar.txt'],
+                     cmd='echo bar > bar.txt')
+
+        self.dvc.run(deps=['bar.txt'],
+                     outs=['baz.txt'],
+                     cmd='echo baz > baz.txt')
+
+        with self.assertRaises(CyclicGraphError):
+            self.dvc.run(deps=['baz.txt'],
+                         outs=[self.FOO],
+                         cmd='echo baz > foo')
+
 
 class TestRunDuplicatedArguments(TestDvc):
     def test(self):

pip install -e .[all]
parse_log_pytest
requirements.txt
pytest>=3.3.1 nose>=1.3.7 mock>=2.0.0 coverage>=4.5.1 codecov>=2.0.15 xmltodict>=0.11.0 awscli>=1.14.18 google-compute-engine Pygments collective.checkdocs
apt-get update apt-get install -y gcc
3.6
requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
altgraph==0.17.4
appdirs==1.4.4
asciimatics==1.14.0
attrs==22.2.0
awscli==1.24.10
azure-common==1.1.28
azure-nspkg==3.0.2
azure-storage-blob==1.3.0
azure-storage-common==1.3.0
azure-storage-nspkg==3.1.0
bcrypt==4.0.1
boto==2.49.0
boto3==1.7.4
botocore==1.10.84
cachetools==4.2.4
certifi==2021.5.30
cffi==1.15.1
charset-normalizer==2.0.12
codecov==2.1.13
collective.checkdocs==0.2
colorama==0.4.4
configobj==5.0.8
configparser==5.2.0
coverage==6.2
cryptography==40.0.2
decorator==4.4.2
distro==1.9.0
docutils==0.16
-e git+https://github.com/iterative/dvc.git@7ba4aeae4236d49430326760f6a6bf52bd714a5f#egg=dvc
future==1.0.0
gitdb==4.0.9
GitPython==3.1.18
google-api-core==1.32.0
google-auth==1.35.0
google-cloud-core==0.28.1
google-cloud-storage==1.13.0
google-compute-engine==2.8.13
google-crc32c==1.3.0
google-resumable-media==2.3.3
googleapis-common-protos==1.56.3
grandalf==0.6
idna==3.10
importlib-metadata==4.8.3
iniconfig==1.1.1
jmespath==0.10.0
jsonpath-rw==1.4.0
macholib==1.16.3
mock==5.2.0
nanotime==0.5.2
networkx==2.5.1
nose==1.3.7
packaging==21.3
paramiko==3.5.1
pefile==2024.8.26
Pillow==8.4.0
pluggy==1.0.0
ply==3.11
protobuf==3.19.6
py==1.11.0
pyasn1==0.5.1
pyasn1-modules==0.3.0
pycparser==2.21
pydot==1.4.2
pyfiglet==0.8.post1
Pygments==2.14.0
PyInstaller==3.3.1
PyNaCl==1.5.0
pyparsing==3.1.4
pytest==7.0.1
python-dateutil==2.9.0.post0
pytz==2025.2
PyYAML==5.4.1
requests==2.27.1
rsa==4.7.2
s3transfer==0.1.13
schema==0.7.7
six==1.17.0
smmap==5.0.0
tomli==1.2.3
typing_extensions==4.1.1
urllib3==1.26.20
wcwidth==0.2.13
xmltodict==0.14.2
zc.lockfile==2.0
zipp==3.6.0

name: dvc
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - altgraph==0.17.4
      - appdirs==1.4.4
      - asciimatics==1.14.0
      - attrs==22.2.0
      - awscli==1.24.10
      - azure-common==1.1.28
      - azure-nspkg==3.0.2
      - azure-storage-blob==1.3.0
      - azure-storage-common==1.3.0
      - azure-storage-nspkg==3.1.0
      - bcrypt==4.0.1
      - boto==2.49.0
      - boto3==1.7.4
      - botocore==1.10.84
      - cachetools==4.2.4
      - cffi==1.15.1
      - charset-normalizer==2.0.12
      - codecov==2.1.13
      - collective-checkdocs==0.2
      - colorama==0.4.4
      - configobj==5.0.8
      - configparser==5.2.0
      - coverage==6.2
      - cryptography==40.0.2
      - decorator==4.4.2
      - distro==1.9.0
      - docutils==0.16
      - future==1.0.0
      - gitdb==4.0.9
      - gitpython==3.1.18
      - google-api-core==1.32.0
      - google-auth==1.35.0
      - google-cloud-core==0.28.1
      - google-cloud-storage==1.13.0
      - google-compute-engine==2.8.13
      - google-crc32c==1.3.0
      - google-resumable-media==2.3.3
      - googleapis-common-protos==1.56.3
      - grandalf==0.6
      - idna==3.10
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - jmespath==0.10.0
      - jsonpath-rw==1.4.0
      - macholib==1.16.3
      - mock==5.2.0
      - nanotime==0.5.2
      - networkx==2.5.1
      - nose==1.3.7
      - packaging==21.3
      - paramiko==3.5.1
      - pefile==2024.8.26
      - pillow==8.4.0
      - pluggy==1.0.0
      - ply==3.11
      - protobuf==3.19.6
      - py==1.11.0
      - pyasn1==0.5.1
      - pyasn1-modules==0.3.0
      - pycparser==2.21
      - pydot==1.4.2
      - pyfiglet==0.8.post1
      - pygments==2.14.0
      - pyinstaller==3.3.1
      - pynacl==1.5.0
      - pyparsing==3.1.4
      - pytest==7.0.1
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - pyyaml==5.4.1
      - requests==2.27.1
      - rsa==4.7.2
      - s3transfer==0.1.13
      - schema==0.7.7
      - six==1.17.0
      - smmap==5.0.0
      - tomli==1.2.3
      - typing-extensions==4.1.1
      - urllib3==1.26.20
      - wcwidth==0.2.13
      - xmltodict==0.14.2
      - zc-lockfile==2.0
      - zipp==3.6.0
prefix: /opt/conda/envs/dvc

tests/test_repro.py::TestReproFail::test tests/test_repro.py::TestReproCyclicGraph::test tests/test_repro.py::TestReproWorkingDirectoryAsOutput::test tests/test_repro.py::TestReproWorkingDirectoryAsOutput::test_nested tests/test_repro.py::TestReproWorkingDirectoryAsOutput::test_similar_paths tests/test_repro.py::TestReproDepUnderDir::test tests/test_repro.py::TestReproDepDirWithOutputsUnderIt::test tests/test_repro.py::TestReproNoDeps::test tests/test_repro.py::TestReproForce::test tests/test_repro.py::TestReproChangedCode::test tests/test_repro.py::TestReproChangedData::test tests/test_repro.py::TestReproDry::test tests/test_repro.py::TestReproUpToDate::test tests/test_repro.py::TestReproDryNoExec::test tests/test_repro.py::TestReproChangedDeepData::test tests/test_repro.py::TestReproIgnoreBuildCache::test tests/test_repro.py::TestReproPipeline::test tests/test_repro.py::TestReproPipeline::test_cli tests/test_repro.py::TestReproPipelines::test tests/test_repro.py::TestReproPipelines::test_cli tests/test_repro.py::TestReproLocked::test tests/test_repro.py::TestReproLocked::test_non_existing tests/test_repro.py::TestReproLockedCallback::test tests/test_repro.py::TestReproLockedUnchanged::test tests/test_repro.py::TestReproMetricsAddUnchanged::test tests/test_repro.py::TestReproPhony::test tests/test_repro.py::TestNonExistingOutput::test tests/test_repro.py::TestReproDataSource::test tests/test_repro.py::TestReproChangedDir::test tests/test_repro.py::TestReproChangedDirData::test tests/test_repro.py::TestReproMissingMd5InStageFile::test tests/test_repro.py::TestCmdRepro::test tests/test_repro.py::TestCmdReproChdir::test tests/test_repro.py::TestReproExternalBase::test tests/test_repro.py::TestReproExternalS3::test tests/test_repro.py::TestReproExternalGS::test tests/test_repro.py::TestReproExternalHDFS::test tests/test_repro.py::TestReproExternalSSH::test tests/test_repro.py::TestReproExternalLOCAL::test tests/test_repro.py::TestReproExternalHTTP::test tests/test_repro.py::TestReproNoSCM::test tests/test_repro.py::TestReproAllPipelines::test tests/test_run.py::TestRun::test tests/test_run.py::TestRunEmpty::test tests/test_run.py::TestRunMissingDep::test tests/test_run.py::TestRunBadStageFilename::test tests/test_run.py::TestRunNoExec::test tests/test_run.py::TestRunCircularDependency::test tests/test_run.py::TestRunCircularDependency::test_graph tests/test_run.py::TestRunCircularDependency::test_non_normalized_paths tests/test_run.py::TestRunCircularDependency::test_outs_no_cache tests/test_run.py::TestRunDuplicatedArguments::test tests/test_run.py::TestRunDuplicatedArguments::test_non_normalized_paths tests/test_run.py::TestRunDuplicatedArguments::test_outs_no_cache tests/test_run.py::TestRunWorkingDirectoryAsOutput::test tests/test_run.py::TestRunBadCwd::test tests/test_run.py::TestCmdRun::test_keyboard_interrupt tests/test_run.py::TestCmdRun::test_run tests/test_run.py::TestCmdRun::test_run_args_from_cli tests/test_run.py::TestCmdRun::test_run_args_with_spaces tests/test_run.py::TestCmdRun::test_run_bad_command tests/test_run.py::TestRunRemoveOuts::test tests/test_run.py::TestCmdRunOverwrite::test tests/test_run.py::TestRunDeterministic::test tests/test_run.py::TestRunDeterministicOverwrite::test tests/test_run.py::TestRunDeterministicCallback::test tests/test_run.py::TestRunDeterministicChangedDep::test tests/test_run.py::TestRunDeterministicChangedDepsList::test tests/test_run.py::TestRunDeterministicNewDep::test tests/test_run.py::TestRunDeterministicRemoveDep::test tests/test_run.py::TestRunDeterministicChangedOut::test tests/test_run.py::TestRunDeterministicChangedCmd::test
tests/test_repro.py::TestReproShell::test tests/test_run.py::TestRunUnprotectOutsCopy::test tests/test_run.py::TestRunUnprotectOutsSymlink::test tests/test_run.py::TestRunUnprotectOutsHardlink::test
dvc/stage.py dvc/project.py dvc/exceptions.py
dvc/stage.py dvc/project.py dvc/exceptions.py