jcrist: Planning on merging tomorrow if no comment.
mrocklin: I don't really know what's going on here, but tests pass so I'm happy.  +1 to merge
diff --git a/dask/optimization.py b/dask/optimization.py
index e2a181f76..c3af897ee 100644
--- a/dask/optimization.py
+++ b/dask/optimization.py
@@ -567,7 +567,10 @@ def fuse(dsk, keys=None, dependencies=None, ave_width=None, max_width=None,
     reducible = {k for k, vals in rdeps.items() if len(vals) == 1}
     if keys:
         reducible -= keys
-    if not reducible:
+    if (not reducible and
+            (not fuse_subgraphs or all(len(set(v)) != 1 for v in rdeps.values()))):
+        # Quick return if there's nothing to do. Only progress if there's tasks
+        # fusible by the main `fuse`, or by `fuse_subgraphs` if enabled.
         return dsk, deps
 
     rv = dsk.copy()

Fusing purely linear chains not working
On branch `master`, in particular after https://github.com/dask/dask/pull/3979 was merged: I am trying to make use of the new feature of merging linear chains in (sub)graphs. To do so, I enable it via the config:

In `config.yaml`:
```
fuse_subgraphs: True
```

The following works as expected, with the new `fuse_subgraphs` option the long chains are now fused:

```python
import dask
import dask.array as da
import numpy as np

if __name__ == '__main__':
    x = da.from_array(np.zeros((3,)), chunks=(1,))
    x = x + x
    branch = x + x
    x = x + branch
    x = x + x
    x = x + x
    x = x + x
    x = x + x
    dask.visualize(x, filename='fused.svg', optimize_graph=True)
```

However, graphs without any branches will not fuse:

```python
import dask
import dask.array as da
import numpy as np

if __name__ == '__main__':
    x = da.from_array(np.zeros((3,)), chunks=(1,))
    x = x + x
    x = x + x
    x = x + x
    x = x + x
    x = x + x
    x = x + x
    x = x + x
    dask.visualize(x, filename='not-fused.svg', optimize_graph=True)
```

I have tracked the issue back to https://github.com/dask/dask/blob/master/dask/optimization.py#L570 --- for a linear graph this returns early, skipping fusion of linear chains. I do not understand what exactly is done in this part of the code, so I cannot propose a precise fix.
diff --git a/dask/tests/test_optimization.py b/dask/tests/test_optimization.py
index 304ab8374..df7164357 100644
--- a/dask/tests/test_optimization.py
+++ b/dask/tests/test_optimization.py
@@ -1200,3 +1200,27 @@ def test_fuse_subgraphs():
             'inc-6': (inc, (inc, 'add-2'))
         }))
     assert res in sols
+
+
+def test_fuse_subgraphs_linear_chains_of_duplicate_deps():
+    dsk = {'x-1': 1,
+           'add-1': (add, 'x-1', 'x-1'),
+           'add-2': (add, 'add-1', 'add-1'),
+           'add-3': (add, 'add-2', 'add-2'),
+           'add-4': (add, 'add-3', 'add-3'),
+           'add-5': (add, 'add-4', 'add-4')}
+
+    res = fuse(dsk, 'add-5', fuse_subgraphs=True)
+    sol = with_deps({
+        'add-x-1': (
+            SubgraphCallable({
+                'x-1': 1,
+                'add-1': (add, 'x-1', 'x-1'),
+                'add-2': (add, 'add-1', 'add-1'),
+                'add-3': (add, 'add-2', 'add-2'),
+                'add-4': (add, 'add-3', 'add-3'),
+                'add-5': (add, 'add-4', 'add-4')
+            }, 'add-5', ()),),
+        'add-5': 'add-x-1'
+    })
+    assert res == sol

pip install -e .[complete]
parse_log_pytest
pytest
pytest pytest-cov pytest-xdist pytest-mock pytest-asyncio
3.9
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
click==8.1.8
cloudpickle==3.1.1
coverage==7.8.0
-e git+https://github.com/dask/dask.git@ecc1e806898410784efc3d4cecf1f02ea5ae2de1#egg=dask
distributed==1.28.1
exceptiongroup @ file:///croot/exceptiongroup_1706031385326/work
execnet==2.1.1
iniconfig @ file:///home/linux1/recipes/ci/iniconfig_1610983019677/work
locket==1.0.0
msgpack==1.1.0
numpy==2.0.2
packaging @ file:///croot/packaging_1734472117206/work
pandas==2.2.3
partd==1.4.2
pluggy @ file:///croot/pluggy_1733169602837/work
psutil==7.0.0
pytest @ file:///croot/pytest_1738938843180/work
pytest-asyncio==0.26.0
pytest-cov==6.0.0
pytest-mock==3.14.0
pytest-xdist==3.6.1
python-dateutil==2.9.0.post0
pytz==2025.2
PyYAML==6.0.2
six==1.17.0
sortedcontainers==2.4.0
tblib==3.1.0
tomli @ file:///opt/conda/conda-bld/tomli_1657175507142/work
toolz==1.0.0
tornado==6.4.2
typing_extensions==4.13.0
tzdata==2025.2
zict==3.0.0

name: dask
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
  - exceptiongroup=1.2.0=py39h06a4308_0
  - iniconfig=1.1.1=pyhd3eb1b0_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - 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=3.0.16=h5eee18b_0
  - packaging=24.2=py39h06a4308_0
  - pip=25.0=py39h06a4308_0
  - pluggy=1.5.0=py39h06a4308_0
  - pytest=8.3.4=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - tomli=2.0.1=py39h06a4308_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - click==8.1.8
      - cloudpickle==3.1.1
      - coverage==7.8.0
      - distributed==1.28.1
      - execnet==2.1.1
      - locket==1.0.0
      - msgpack==1.1.0
      - numpy==2.0.2
      - pandas==2.2.3
      - partd==1.4.2
      - psutil==7.0.0
      - pytest-asyncio==0.26.0
      - pytest-cov==6.0.0
      - pytest-mock==3.14.0
      - pytest-xdist==3.6.1
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - pyyaml==6.0.2
      - six==1.17.0
      - sortedcontainers==2.4.0
      - tblib==3.1.0
      - toolz==1.0.0
      - tornado==6.4.2
      - typing-extensions==4.13.0
      - tzdata==2025.2
      - zict==3.0.0
prefix: /opt/conda/envs/dask

dask/tests/test_optimization.py::test_fuse_subgraphs_linear_chains_of_duplicate_deps
dask/tests/test_optimization.py::test_cull dask/tests/test_optimization.py::test_fuse dask/tests/test_optimization.py::test_fuse_keys dask/tests/test_optimization.py::test_inline dask/tests/test_optimization.py::test_inline_functions dask/tests/test_optimization.py::test_inline_ignores_curries_and_partials dask/tests/test_optimization.py::test_inline_functions_non_hashable dask/tests/test_optimization.py::test_inline_doesnt_shrink_fast_functions_at_top dask/tests/test_optimization.py::test_inline_traverses_lists dask/tests/test_optimization.py::test_inline_functions_protects_output_keys dask/tests/test_optimization.py::test_functions_of dask/tests/test_optimization.py::test_fuse_getitem dask/tests/test_optimization.py::test_fuse_selections dask/tests/test_optimization.py::test_inline_cull_dependencies dask/tests/test_optimization.py::test_fuse_reductions_single_input dask/tests/test_optimization.py::test_fuse_stressed dask/tests/test_optimization.py::test_fuse_reductions_multiple_input dask/tests/test_optimization.py::test_SubgraphCallable dask/tests/test_optimization.py::test_fuse_subgraphs
dask/optimization.py
dask/optimization.py