fix(workflows): fan-out max_concurrency .inf falls back to sequential, not crash (#3521)

_run_fan_out coerces max_concurrency with int() inside except (TypeError,
ValueError). int(float('inf')) raises OverflowError, which is not in that tuple,
so a YAML 'max_concurrency: .inf' crashed the whole run with an uncaught
OverflowError instead of the documented 'cannot be coerced -> sequential'
fallback. Add OverflowError to the except tuple (nan already coerced via
ValueError).

Extends the existing invalid-value parametrization with float('inf')/nan (fails
before on inf: OverflowError; passes after: sequential, all items in order).

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ali jawwad
2026-07-15 18:06:03 +05:00
committed by GitHub
parent 1e84ee2713
commit fb076a38b8
2 changed files with 12 additions and 5 deletions

View File

@@ -1197,9 +1197,9 @@ class WorkflowEngine:
already flipped), so the prefix never drops the actual halting item.
``max_concurrency`` is coerced with ``int()``; a value that cannot be
coerced (``None``, a non-numeric string, …) or that coerces to <= 1 runs
sequentially, while a numeric string like ``"4"`` or a float like ``4.0``
is honored.
coerced (``None``, a non-numeric string, ``.inf``/``.nan``, …) or that
coerces to <= 1 runs sequentially, while a numeric string like ``"4"`` or
a float like ``4.0`` is honored.
"""
if not items:
return []
@@ -1207,7 +1207,9 @@ class WorkflowEngine:
halting = (RunStatus.PAUSED, RunStatus.FAILED, RunStatus.ABORTED)
try:
workers = max(1, int(max_concurrency))
except (TypeError, ValueError):
except (TypeError, ValueError, OverflowError):
# OverflowError: int(float("inf")) — a YAML ``max_concurrency: .inf``
# would otherwise crash the whole run instead of falling back.
workers = 1
# Never spin up more workers than there is work — bounds a user-controlled
# max_concurrency from over-allocating threads.

View File

@@ -2727,8 +2727,13 @@ class TestFanOutConcurrency:
results, _ = self._run(tmp_path, list(range(n)), n, on_item)
assert results == [{"seen": i} for i in range(n)]
@pytest.mark.parametrize("bad", [0, -1, None, "abc", 1.0])
@pytest.mark.parametrize(
"bad", [0, -1, None, "abc", 1.0, float("inf"), float("nan")]
)
def test_invalid_max_concurrency_coerces_to_sequential(self, tmp_path, bad):
# float("inf") -> int() raises OverflowError (not TypeError/ValueError);
# it must fall back to sequential like any other uncoercible value, not
# crash the run.
results, _ = self._run(tmp_path, list(range(4)), bad)
assert results == [{"seen": i} for i in range(4)]