Repository navigation
feat(mlops): Add run conditions to step dependencies - #6475
Conversation
Add StepDependency and TriggerCondition so a step can run when an upstream step fails, stops, or reaches any terminal state. Completed dependencies are still emitted as plain step names, so existing pipeline definitions are unchanged. Local mode rejects non-Completed triggers because it stops on the first failed step. Integ test passed against a production region. --- X-AI-Prompt: Add backward compatible Run Conditions support to the PySDK and raise a PR against master X-AI-Tool: kiro-cli
GitHub Actions
Error
GitHub Actions
Error
GitHub Actions
Error
| dependencies = {} | ||
| for item in step.depends_on or []: | ||
| if not isinstance(item, StepDependency): | ||
| continue |
There was a problem hiding this comment.
Can we make sure we have unit tests for all possible combinations here of mixed StepDependency + str and pure types? I think some cases are not raising "conflicting triggers" when they should
depends_on= |
emitted DependsOn |
|
|---|---|---|
["A"] |
['A'] |
expected |
[a] |
['A'] |
expected |
[a, "A"] |
['A'] |
expected |
[SD(a)] |
['A'] |
expected |
[SD(a, COMPLETED)] |
['A'] |
expected |
[SD(a, FAILED)] |
[{'Step':'A','Trigger':'Failed'}] |
expected |
[SD(a, ANY)] |
[{'Step':'A','Trigger':'Any'}] |
expected |
[SD(a,COMPLETED), SD(a,FAILED)] |
RAISES "conflicting triggers" | expected |
[SD(a,FAILED), SD("A",FAILED)] |
[{'Step':'A','Trigger':'Failed'}] |
expected |
["A", SD(a,FAILED)] |
[{'Step':'A','Trigger':'Failed'}] |
incorrect? |
[a, SD(a,FAILED)] |
[{'Step':'A','Trigger':'Failed'}] |
incorrect? |
[SD(a,FAILED), "A"] |
[{'Step':'A','Trigger':'Failed'}] |
incorrect? |
There was a problem hiding this comment.
Good catch. Plain entries are implicit Completed dependencies, but the conflict check skipped them, so ["A", SD(a, FAILED)] silently emitted Failed. Plain entries are now normalized to StepDependency(item) before the check, so those three cases raise conflicting triggers. Added parametrized tests covering every row in your table.
A plain step or step name listed next to a StepDependency on the same step with a non-Completed trigger was ignored by the conflict check, so the Completed dependency was silently dropped. Plain entries now count as Completed and such combinations raise. --- X-AI-Prompt: Address jkasiraj review on PR 6475 about mixed depends_on conflict cases X-AI-Tool: kiro-cli
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
🤖 AI code review
This PR adds StepDependency and TriggerCondition so a step's depends_on entry can be wired to run on an upstream step's failure, stop, or any terminal state. Completed dependencies are still emitted as plain step names, keeping existing pipeline definitions byte-identical, and local mode rejects non-Completed triggers. The implementation threads the new types cleanly through the compiler (_to_depends_on_request), Step._find_dependencies_in_depends_on_list, and the local executor, and is backed by thorough unit tests plus an integ test. Overall this looks correct and well-tested; I only have minor observations.
Reviewed commit 184a4ac. Automated review; verify before acting.
| """ | ||
| if not isinstance(trigger, TriggerCondition): | ||
| raise TypeError(f"trigger must be a TriggerCondition, got {type(trigger).__name__}.") | ||
| self.step = get_step(step) if isinstance(step, StepOutput) else step |
There was a problem hiding this comment.
🟡 Low: step_name returns self.step.name for the wrapped object. The base Step.depends_on type hint and _find_dependencies_in_depends_on_list both accept StepCollection, and for a StepCollection the real dependency target is steps[-1].name, not the collection's name. If someone wraps a StepCollection in a StepDependency, step_name/to_request would emit the collection name and produce an inconsistent DependsOn entry. The __init__ docstring only advertises Union[str, Step, StepOutput], so this is likely intentional scope — but consider either rejecting StepCollection explicitly or resolving it to steps[-1].name to avoid a silently wrong edge.
|
|
||
| return list(depends_on), upstream_step_in_args | ||
|
|
||
| def _to_depends_on_request(self, step: Step, depends_on: List[Step]) -> List[Union[str, dict]]: |
There was a problem hiding this comment.
🟡 Low: _to_depends_on_request calls self._simplify_step_list([item])[0].name once per entry in step.depends_on, giving O(n^2) resolution over the dependency list. Dependency lists are small so this is not a real performance concern, but resolving names into the dependencies map in a single pass (and only once per unique item) would be a touch cleaner.
This branch was successfully deployed
3 participants
Add StepDependency and TriggerCondition so a step can run when an upstream step fails, stops, or reaches any terminal state. Completed dependencies are still emitted as plain step names, so existing pipeline definitions are unchanged. Local mode rejects non-Completed triggers because it stops on the first failed step.
Integ test passed against a production region.
X-AI-Prompt: Add backward compatible Run Conditions support to the PySDK and raise a PR against master
X-AI-Tool: kiro-cli
Issue #, if available:
Description of changes:
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.