Skip to content

Commit 2aa0f4b

Browse files
fix(workflows): publish dynamic nested descendants from isolated fan-out items
An isolated (concurrent, or nested-in-isolated) fan-out item published back to the shared steps dict only the ids in its static id_map. Ids generated at runtime by a nested while/do-while body or fan-out (e.g. fan:inner:0:leaf:0) reached state.step_results but not the live context.steps, so a later `steps.<id>` reference resolved to nothing on a fresh concurrent run. Publish every key the item added to or replaced in its private view, excluding its bare-id aliases (still folded in item order by the caller). The resume-path skip and run-lock behavior are unchanged. Adds a container x concurrency x fresh/resume regression matrix. Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent f5088bd commit 2aa0f4b

2 files changed

Lines changed: 134 additions & 15 deletions

File tree

‎src/specify_cli/workflows/engine.py‎

Lines changed: 23 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1858,10 +1858,9 @@ def run_item(
18581858
# the shared dict made after this item started, but fan-out
18591859
# items were never entitled to see those anyway.
18601860
original_steps = item_ctx.steps
1861+
snapshot_src = original_steps if base_steps is None else base_steps
18611862
if local_only:
1862-
item_steps = dict(
1863-
original_steps if base_steps is None else base_steps
1864-
)
1863+
item_steps = dict(snapshot_src)
18651864
else:
18661865
item_steps = original_steps
18671866
item_ctx.steps = item_steps
@@ -1914,18 +1913,27 @@ def run_item(
19141913
original_steps[orig] = val
19151914
steps_view = item_steps if local_only else item_ctx.steps
19161915
if local_only and original_steps is not state.step_results:
1917-
for new_id in id_map:
1918-
if new_id in steps_view:
1919-
# Publish the namespaced (disjoint, per-item) result
1920-
# into the truly-shared steps dict explicitly — safe
1921-
# even under concurrency since each item only ever
1922-
# writes its own namespaced keys here. Skipped when
1923-
# that dict IS ``state.step_results`` (a resume run —
1924-
# see ``_record_result``): _record_result already
1925-
# wrote every namespaced key there under the run
1926-
# lock, and an unlocked write here could race another
1927-
# worker's ``state.save()`` iterating it.
1928-
original_steps[new_id] = steps_view[new_id]
1916+
# Publish every namespaced (disjoint, per-item) result this
1917+
# item recorded into the truly-shared steps dict explicitly —
1918+
# safe even under concurrency since each item only ever
1919+
# writes its own namespaced keys here. That is every key the
1920+
# item added to or replaced in its private view, minus its
1921+
# bare-id aliases (``alias_records``), which the caller
1922+
# publishes in item order instead. Diffing the view rather
1923+
# than walking ``id_map`` also catches ids a nested
1924+
# while/do-while body or fan-out generated dynamically at
1925+
# runtime, which the static ``id_map`` never contains.
1926+
# Skipped when that dict IS ``state.step_results`` (a resume
1927+
# run — see ``_record_result``): _record_result already wrote
1928+
# every namespaced key there under the run lock, and an
1929+
# unlocked write here could race another worker's
1930+
# ``state.save()`` iterating it.
1931+
for key, val in steps_view.items():
1932+
if key in alias_records:
1933+
continue
1934+
if snapshot_src.get(key, _MISSING_STEP) is val:
1935+
continue
1936+
original_steps[key] = val
19291937
# Read back through the local view, not the outer closure —
19301938
# clearer and robust if StepContext copying ever stops sharing
19311939
# the steps dict by reference.

‎tests/test_workflows.py‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7948,6 +7948,117 @@ def test_fan_out_reserved_id_collision_end_to_end_on_resume(
79487948
)
79497949
assert state.step_results["after"]["output"]["stdout"].strip() == "outside"
79507950

7951+
@pytest.mark.parametrize("resume", [False, True], ids=["fresh", "resume"])
7952+
@pytest.mark.parametrize("max_concurrency", [1, 2])
7953+
@pytest.mark.parametrize("container", ["fan-out", "while", "do-while"])
7954+
def test_nested_dynamic_descendants_visible_downstream(
7955+
self, project_dir, container, max_concurrency, resume
7956+
):
7957+
"""Every namespaced result an outer fan-out item produces must reach
7958+
the live `context.steps`, including ids a nested while/do-while
7959+
body or fan-out generates dynamically at runtime.
7960+
7961+
A concurrent item publishes its private steps view back to the
7962+
shared dict after it finishes. That publish used to walk only the
7963+
item's static `id_map`, which never contains a nested container's
7964+
runtime-generated ids (`fan:inner:<i>:leaf:0`). Those results
7965+
still reached `state.step_results`, but a later `steps.<id>`
7966+
reference resolved to nothing on a fresh concurrent run.
7967+
"""
7968+
from unittest.mock import patch
7969+
7970+
from specify_cli.workflows.base import RunStatus, StepResult
7971+
from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine
7972+
7973+
inner = {
7974+
"fan-out": """\
7975+
id: inner
7976+
type: fan-out
7977+
items: "{{ ['x'] }}"
7978+
max_concurrency: 1
7979+
step:
7980+
id: leaf
7981+
type: shell
7982+
run: "echo {{ item }}-{{ item }}"
7983+
""",
7984+
"while": """\
7985+
id: inner
7986+
type: while
7987+
condition: "true"
7988+
max_iterations: 1
7989+
steps:
7990+
- id: leaf
7991+
type: shell
7992+
run: "echo x-x"
7993+
""",
7994+
"do-while": """\
7995+
id: inner
7996+
type: do-while
7997+
condition: "false"
7998+
steps:
7999+
- id: leaf
8000+
type: shell
8001+
run: "echo x-x"
8002+
""",
8003+
}[container]
8004+
gate = (
8005+
"""\
8006+
- id: approve
8007+
type: gate
8008+
message: "Approve?"
8009+
"""
8010+
if resume
8011+
else ""
8012+
)
8013+
yaml_str = (
8014+
"""
8015+
schema_version: "1.0"
8016+
workflow:
8017+
id: "fan-out-nested-dynamic-visibility"
8018+
name: "Fan Out Nested Dynamic Visibility"
8019+
version: "1.0.0"
8020+
steps:
8021+
"""
8022+
+ gate
8023+
+ f"""\
8024+
- id: fan
8025+
type: fan-out
8026+
items: "{{{{ ['a', 'b'] }}}}"
8027+
max_concurrency: {max_concurrency}
8028+
step:
8029+
"""
8030+
+ inner
8031+
+ "".join(
8032+
f"""\
8033+
- id: after{idx}
8034+
type: shell
8035+
run: "echo {{{{ steps.fan:inner:{idx}:leaf:0.output.stdout | default('MISSING') }}}}"
8036+
"""
8037+
for idx in range(2)
8038+
)
8039+
)
8040+
definition = WorkflowDefinition.from_string(yaml_str)
8041+
engine = WorkflowEngine(project_dir)
8042+
state = engine.execute(definition)
8043+
if resume:
8044+
assert state.status == RunStatus.PAUSED
8045+
with patch(
8046+
"specify_cli.workflows.step.gate.GateStep.execute",
8047+
return_value=StepResult(output={"approved": True}),
8048+
):
8049+
state = engine.resume(state.run_id)
8050+
8051+
assert state.status == RunStatus.COMPLETED
8052+
for idx in range(2):
8053+
assert (
8054+
state.step_results[f"fan:inner:{idx}:leaf:0"]["output"]["stdout"]
8055+
== "x-x\n"
8056+
)
8057+
assert (
8058+
state.step_results[f"after{idx}"]["output"]["stdout"].strip()
8059+
== "x-x"
8060+
)
8061+
79518062
def test_concurrent_fan_out_items_copy_fixed_pre_fan_out_snapshot(
79528063
self, tmp_path
79538064
):

0 commit comments

Comments
 (0)