One failed step in an AI workflow can block several business outputs. Before retrying everything or blaming the last step, map the mandatory dependencies and calculate which tasks are downstream of the failure.
This guide uses NetworkX to turn a small task map into dependency layers and a failure-impact report. The example rejects cycles and misspelled prerequisites before producing a plan. It analyzes a declared workflow; it does not execute an agent, approve a quote or send a message.
Draw the dependencies before writing the retry
Harbor Spark is a fictional business planning an AI-assisted quote workflow. Its authored map has seven tasks. Drafting needs both the request and a validated catalog. Human review follows the draft, and sending follows review. Archiving the request is a separate branch.
load_catalog -> validate_catalog -> draft_quote -> human_review -> send_quote
^
|
read_request --------------------------+
|
+------> archive_requestIf validate_catalog is declared failed, draft_quote, human_review and send_quote are blocked descendants. archive_request has no dependency on that failure. That does not make archiving automatically ready or authorized: the report has no completion ledger, permission model or live input state.
Make the arrow direction explicit
The input maps each task to its prerequisites. The graph stores the reverse relationship as an edge: prerequisite to dependent. Reversing that edge would answer a different question and can produce a plausible but wrong impact report.
NetworkX’s descendants operation returns reachable downstream nodes and excludes the source. The example keeps failed tasks in their own list, unions their descendants and removes any tasks already classified as failed. Multiple failures therefore do not duplicate a blocked task or relabel a declared failure as merely blocked.
The topological generations API groups a directed acyclic graph into dependency layers. Its iterator can raise a cycle error during iteration. The example consumes the complete iterator before returning any report, so a cyclic graph cannot slip through as a partial plan.
Use the tested package version
Execution used Windows, CPython 3.14.3 and the already installed NetworkX 3.6.1, with no configured algorithm backend priority. The current stable API pages identify themselves as 3.7; that newer package was not tested. Reproduce the recorded environment with NetworkX==3.6.1 in your own Python environment. The execution here reused the existing installation; a clean package installation was not exercised.
The exact 3.6.1 DAG source was inspected for the traversal and layer logic. Its project metadata declares Python 3.11 or newer, excluding 3.14.1, and no base runtime dependencies; optional extras are separate. Its BSD three-clause license carries redistribution conditions. No repository installer or paid service was used.
Run the analysis and failure tests
Save the complete example as dependency_plan.py in a new disposable directory. Task IDs are short lowercase identifiers; dependency and failure lists must refer to declared tasks. The example caps the graph at 100 tasks and each dependency list at 100 entries. Those are authored example limits, not a benchmark or a NetworkX capacity claim.
"""Static dependency analysis only: no task execution or external actions."""
import copy
import json
import re
import unittest
import networkx as nx
TASKS = {
'read_request': [],
'load_catalog': [],
'validate_catalog': ['load_catalog'],
'draft_quote': ['read_request', 'validate_catalog'],
'human_review': ['draft_quote'],
'send_quote': ['human_review'],
'archive_request': ['read_request'],
}
def analyze(tasks, failed):
if not isinstance(tasks, dict) or not 1 <= len(tasks) <= 100:
raise ValueError('Use a mapping of 1 to 100 tasks')
if any(not isinstance(k, str) or re.fullmatch(r'[a-z][a-z0-9_]{0,39}', k) is None
for k in tasks):
raise ValueError('Use short lowercase task IDs')
graph = nx.DiGraph()
graph.add_nodes_from(tasks)
for task, dependencies in tasks.items():
if not isinstance(dependencies, list) or len(dependencies) > 100:
raise ValueError('Dependencies must be lists of at most 100 IDs')
if any(not isinstance(d, str) or d not in tasks for d in dependencies):
raise ValueError('Every dependency must name a declared task')
if len(set(dependencies)) != len(dependencies):
raise ValueError('Repeated dependency')
# prerequisite -> dependent, never the reverse
graph.add_edges_from((dependency, task) for dependency in dependencies)
try:
layers = [sorted(layer) for layer in nx.topological_generations(graph)]
except nx.NetworkXUnfeasible as error:
raise ValueError('Dependency cycle: no complete plan exists') from error
if not isinstance(failed, list) or len(failed) > len(tasks):
raise ValueError('Failed IDs must be a bounded list')
if any(not isinstance(f, str) or f not in tasks for f in failed):
raise ValueError('Failed ID is not a declared task')
failed_set = set(failed)
if len(failed_set) != len(failed):
raise ValueError('Repeated failed ID')
blocked = set()
for task in failed_set:
blocked.update(nx.descendants(graph, task))
blocked -= failed_set
return {
'dependency_layers': layers,
'failed': sorted(failed_set),
'blocked_descendants': sorted(blocked),
'unaffected_by_declared_failures': sorted(set(tasks) - failed_set - blocked),
}
class PlanTests(unittest.TestCase):
def test_expected_layers(self):
self.assertEqual(analyze(TASKS, [])['dependency_layers'], [
['load_catalog', 'read_request'], ['archive_request', 'validate_catalog'],
['draft_quote'], ['human_review'], ['send_quote']])
def test_catalog_failure_blocks_only_descendants(self):
result = analyze(TASKS, ['validate_catalog'])
self.assertEqual(result['blocked_descendants'], ['draft_quote', 'human_review', 'send_quote'])
self.assertEqual(result['unaffected_by_declared_failures'], ['archive_request', 'load_catalog', 'read_request'])
def test_leaf_failure_blocks_nothing_else(self):
self.assertEqual(analyze(TASKS, ['send_quote'])['blocked_descendants'], [])
def test_overlapping_failures_are_not_double_counted(self):
result = analyze(TASKS, ['validate_catalog', 'human_review'])
self.assertEqual(result['failed'], ['human_review', 'validate_catalog'])
self.assertEqual(result['blocked_descendants'], ['draft_quote', 'send_quote'])
def test_unknown_dependency_is_rejected(self):
with self.assertRaises(ValueError): analyze({'draft': ['typo']}, [])
def test_cycles_are_rejected(self):
for tasks in [{'a': ['b'], 'b': ['a']}, {'a': ['a']}]:
with self.subTest(tasks=tasks), self.assertRaises(ValueError): analyze(tasks, [])
def test_mapping_size_and_type(self):
for tasks in [{}, [], {f't{i}': [] for i in range(101)}]:
with self.subTest(size=len(tasks)), self.assertRaises(ValueError): analyze(tasks, [])
def test_dependency_type_and_duplicates(self):
for dependencies in ['a', [7], ['a', 'a']]:
with self.subTest(dependencies=dependencies), self.assertRaises(ValueError):
analyze({'a': [], 'b': dependencies}, [])
def test_invalid_task_ids(self):
for task in [7, 'Upper', 'space id', 'a'*41]:
with self.subTest(task=task), self.assertRaises(ValueError): analyze({task: []}, [])
def test_failed_input_is_checked(self):
for failed in ['read_request', [7], ['typo'], ['read_request', 'read_request']]:
with self.subTest(failed=failed), self.assertRaises(ValueError): analyze(TASKS, failed)
def test_input_is_unchanged(self):
before = copy.deepcopy(TASKS)
analyze(TASKS, ['validate_catalog'])
self.assertEqual(TASKS, before)
def test_disconnected_task_is_retained(self):
result = analyze({'a': [], 'b': []}, ['a'])
self.assertEqual(result['unaffected_by_declared_failures'], ['b'])
self.assertEqual(result['dependency_layers'], [['a', 'b']])
if __name__ == '__main__':
print(json.dumps(analyze(TASKS, ['validate_catalog']), indent=2))
Run both commands. The first reports twelve passing test methods. The second analyzes the fictional map with validate_catalog declared failed.
python -X utf8 -m unittest -v dependency_plan.py
python -X utf8 dependency_plan.py{
"dependency_layers": [
[
"load_catalog",
"read_request"
],
[
"archive_request",
"validate_catalog"
],
[
"draft_quote"
],
[
"human_review"
],
[
"send_quote"
]
],
"failed": [
"validate_catalog"
],
"blocked_descendants": [
"draft_quote",
"human_review",
"send_quote"
],
"unaffected_by_declared_failures": [
"archive_request",
"load_catalog",
"read_request"
]
}Read the report without turning it into a scheduler
dependency_layers describes the entire valid graph before considering failures. The two root tasks share layer zero; archiving and catalog validation share layer one. That is structural ordering, not a measured schedule. Shared credentials, rate limits, locks, costs or human availability may still prevent tasks in one layer from running together.
unaffected_by_declared_failures means only that the listed task is neither a declared failure nor reachable downstream from one. It does not mean succeeded, ready now, safe to run or unaffected by an undeclared real-world dependency. In this sample, loading the catalog is upstream of validation; it remains in that list even if its existing output caused the validation failure.
The failed list is supplied by the caller. The analyzer does not discover failures, inspect task outputs or decide whether a cached artifact is usable. Bind those facts to your actual orchestrator’s records before making operational decisions. A stale hand-maintained map can be worse than an explicit unknown.
Reject broken maps instead of guessing
An unknown prerequisite is rejected before adding edges, preventing a typo from quietly becoming a new graph node. Cycles and self-dependencies fail the complete plan. A real review-and-revise loop needs an explicit iteration or state-machine design; remove a cycle only when the underlying workflow has actually been clarified.
This model treats every edge as a mandatory dependency. Optional branches, “either source is sufficient,” fallbacks and retry attempts need richer semantics. Do not infer those rules from reachability alone. The code does not execute callbacks or interpret task IDs as commands.
After a repair, validate the affected artifacts and approvals using the real workflow before resuming downstream work. Recomputing this report with a shorter failed list is not proof of recovery. The separate retryable CRM handoff discusses delivery behavior, while run-scoped logs help keep overlapping investigations tied to the right job.
What the double-check established
Twelve executed test methods cover expected layers, a validation failure’s descendants, leaf failure, overlapping failures, unknown prerequisites, cycles including self-cycles, input size/type, dependency type/duplicates, invalid task IDs, invalid failure lists, input preservation and a disconnected task. The final article code is extracted into a disposable directory and rerun. After publication, public code is compared with independently preserved hashes and executed again.
The confirmed result is a small authored dependency analysis on the recorded runtime. No AI model, agent host, quote calculation, human approval system, queue, scheduler, CRM, email adapter or production application was exercised. There is no claim of throughput, optimal scheduling, workflow recovery or business savings.
Jason Ead publishes AlgoThoughts AI News and connects practical engineering with My Biz Heroes: websites, automation, workflows, AI tools and training. Explore Jason’s work. The astronaut-and-domino cover is generated conceptual artwork; the interior diagram depicts the declared fixture and the output records its executed analysis.