networkx 3.6.1: DiGraph, is_directed_acyclic_graph, topological_sort

# NetworkX Directed Acyclic Graph (DAG) Analysis and Topological Sorting

## Prior Solution Found
`search_known_solution` returned `NO_SAFE_MATCH` for DAG algorithms and topological sorting in `pkg:pypi/networkx@3.6.1`.

## What a Model Would Have Written
A model writing DAG analysis code often:
1. Assumes `nx.topological_sort(G)` returns a concrete list; in NetworkX 2.x and 3.x, `topological_sort` returns an iterator / generator, which must be converted via `list()` or iterated over.
2. Expects `nx.topological_sort(G)` on a cyclic graph to raise `ValueError` or return partial results, rather than catching the specific `nx.NetworkXUnfeasible` exception.
3. Expects `nx.find_cycle(G)` to return empty list or None on acyclic graphs, rather than raising `nx.NetworkXNoCycle`.
4. Confuses `nx.dag_longest_path` (which only works on DAGs and raises `NetworkXNotImplemented` or `NetworkXUnfeasible` on cyclic graphs) with general longest path heuristics.
5. Uses custom graph traversal to find ancestors or descendants instead of using `nx.ancestors(G, source)` and `nx.descendants(G, source)`, which return Python `set` of reachable nodes without including the source node itself.
6. Does not know that `nx.transitive_reduction(G)` strips redundant shortcut edges while preserving reachability, which is essential for minimal dependency graphs.

## How the Wrong Version Fails
- Calling `len(nx.topological_sort(G))` or indexing into `nx.topological_sort(G)` raises `TypeError: object of type 'generator' has no len()` or `TypeError: 'generator' object is not subscriptable`.
- Trying to topologically sort a graph with cycles without catching `nx.NetworkXUnfeasible` crashes pipeline execution.
- Calling `nx.find_cycle(G)` on an acyclic graph without catching `nx.NetworkXNoCycle` causes unhandled exceptions.
- Assuming `nx.ancestors(G, node)` includes `node` leads to off-by-one errors in dependency resolution sets.
- Attempting `nx.dag_longest_path` on graphs with cycles raises `nx.NetworkXUnfeasible`.

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify networkx in pkg:pypi/networkx@3.6.1
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/networkx@3.6.1
Demonstrate these symbols/APIs:
  - networkx

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:b0bf84818bb3c108496d04c3055ed4f24c0380f54d402c606d48e4c536b9261d","contract":["assert is_directed_acyclic_graph accurately determines whether a directed graph is acyclic","assert topological_sort returns an iterator yielding valid topological orderings where dependencies precede dependents","assert topological_sort raises NetworkXUnfeasible when applied to a directed graph with cycles","assert find_cycle locates cycle edges on cyclic graphs and raises NetworkXNoCycle on acyclic graphs","assert ancestors and descendants retrieve complete upstream and downstream node sets without self-inclusion","assert dag_longest_path and dag_longest_path_length compute the critical execution path and distance on unweighted and weighted graphs","assert transitive_reduction prunes transitive shortcut edges while preserving reachability"],"goal":"verify networkx in pkg:pypi/networkx@3.6.1","kind":"HOW","packages":["pkg:pypi/networkx@3.6.1"],"schemaVersion":1,"symbols":["networkx","networkx.DiGraph","networkx.is_directed_acyclic_graph","networkx.topological_sort","networkx.find_cycle","networkx.ancestors","networkx.descendants","networkx.dag_longest_path","networkx.dag_longest_path_length","networkx.transitive_reduction","networkx.NetworkXUnfeasible","networkx.NetworkXNoCycle"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","executionContext":"python","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/networkx@3.6.1"],"schemaVersion":1,"subject":"pkg:pypi/networkx@3.6.1","symbols":["networkx","networkx.DiGraph","networkx.is_directed_acyclic_graph","networkx.topological_sort","networkx.find_cycle","networkx.ancestors","networkx.descendants","networkx.dag_longest_path","networkx.dag_longest_path_length","networkx.transitive_reduction","networkx.NetworkXUnfeasible","networkx.NetworkXNoCycle"],"verifierAdapter":"python@1"}

networkx==3.6.1

{
  "schemaVersion": 1,
  "goal": "verify networkx in pkg:pypi/networkx@3.6.1",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/networkx@3.6.1"
  ],
  "symbols": [
    "networkx"
  ]
}

"""NetworkX directed acyclic graph (DAG) analytics and topological sorting."""

"""Directed acyclic graph (DAG) operations and topological sorting utilities using NetworkX."""

from typing import Any, Dict, List, Optional, Set, Tuple
import networkx as nx


def build_dependency_graph(
    edges: List[Tuple[Any, Any]],
    weights: Optional[Dict[Tuple[Any, Any], float]] = None,
) -> nx.DiGraph:
    """Build a directed graph with optional edge weights."""
    g = nx.DiGraph()
    g.add_edges_from(edges)
    if weights:
        for (u, v), w in weights.items():
            g[u][v]["weight"] = w
    return g


def is_valid_dag(g: nx.DiGraph) -> bool:
    """Check whether a directed graph is acyclic."""
    return nx.is_directed_acyclic_graph(g)


def compute_topological_sort(g: nx.DiGraph) -> List[Any]:
    """Compute topological sorting order of a DAG.

    Raises nx.NetworkXUnfeasible if the graph contains cycles.
    """
    return list(nx.topological_sort(g))


def detect_cycle_or_none(g: nx.DiGraph) -> Optional[List[Tuple[Any, Any]]]:
    """Detect a cycle in the directed graph, returning edge tuples or None if acyclic."""
    try:
        return list(nx.find_cycle(g))
    except nx.NetworkXNoCycle:
        return None


def get_dependency_sets(g: nx.DiGraph, node: Any) -> Dict[str, Set[Any]]:
    """Return transitive upstream ancestors and downstream descendants for a node."""
    return {
        "ancestors": nx.ancestors(g, node),
        "descendants": nx.descendants(g, node),
    }


def compute_critical_path(
    g: nx.DiGraph,
    weight: Optional[str] = None,
) -> Tuple[List[Any], float]:
    """Compute the longest path (critical path) and its total length in a DAG."""
    path = nx.dag_longest_path(g, weight=weight)
    length = float(nx.dag_longest_path_length(g, weight=weight))
    return path, length


def simplify_transitive_dependencies(g: nx.DiGraph) -> nx.DiGraph:
    """Prune redundant shortcut edges while preserving transitive reachability."""
    return nx.transitive_reduction(g)

"""Contract verification suite for pkg:pypi/networkx@3.6.1 DAG and topological sort."""

from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import networkx as nx
from src import dag_ops

# Verify version
assert nx.__version__ == "3.6.1", f"Expected networkx 3.6.1, got {nx.__version__}"

# ============================================================================
# 1. Directed Acyclic Graph Validation
# ============================================================================
pipeline_edges = [
    ("fetch", "build"),
    ("build", "lint"),
    ("build", "test"),
    ("lint", "deploy"),
    ("test", "deploy"),
]
dag = dag_ops.build_dependency_graph(pipeline_edges)
assert dag_ops.is_valid_dag(dag) is True
assert nx.is_directed_acyclic_graph(dag) is True

cyclic_edges = [("A", "B"), ("B", "C"), ("C", "A")]
cyclic_g = dag_ops.build_dependency_graph(cyclic_edges)
assert dag_ops.is_valid_dag(cyclic_g) is False
assert nx.is_directed_acyclic_graph(cyclic_g) is False

# ============================================================================
# 2. Topological Sort Order & Dependency Precedence
# ============================================================================
topo_order = dag_ops.compute_topological_sort(dag)
assert len(topo_order) == len(dag.nodes())

# Ensure topological order satisfies all precedence constraints
node_order_idx = {node: idx for idx, node in enumerate(topo_order)}
for u, v in dag.edges():
    assert node_order_idx[u] < node_order_idx[v], f"Dependency violation: {u} must precede {v}"

# Confirm start and terminal nodes
assert topo_order[0] == "fetch"
assert topo_order[-1] == "deploy"

# ============================================================================
# 3. Topological Sort Raises NetworkXUnfeasible on Cyclic Graphs
# ============================================================================
try:
    dag_ops.compute_topological_sort(cyclic_g)
    assert False, "topological_sort on cyclic graph must raise NetworkXUnfeasible"
except nx.NetworkXUnfeasible:
    pass

# ============================================================================
# 4. Cycle Detection & NetworkXNoCycle Exception
# ============================================================================
cycle = dag_ops.detect_cycle_or_none(cyclic_g)
assert cycle == [("A", "B"), ("B", "C"), ("C", "A")]

no_cycle = dag_ops.detect_cycle_or_none(dag)
assert no_cycle is None

try:
    nx.find_cycle(dag)
    assert False, "find_cycle on acyclic graph must raise NetworkXNoCycle"
except nx.NetworkXNoCycle:
    pass

# ============================================================================
# 5. Transitive Ancestors & Descendants Without Self-Inclusion
# ============================================================================
deploy_deps = dag_ops.get_dependency_sets(dag, "deploy")
assert deploy_deps["ancestors"] == {"fetch", "build", "lint", "test"}
assert "deploy" not in deploy_deps["ancestors"]
assert deploy_deps["descendants"] == set()

fetch_deps = dag_ops.get_dependency_sets(dag, "fetch")
assert fetch_deps["ancestors"] == set()
assert fetch_deps["descendants"] == {"build", "lint", "test", "deploy"}
assert "fetch" not in fetch_deps["descendants"]

build_deps = dag_ops.get_dependency_sets(dag, "build")
assert build_deps["ancestors"] == {"fetch"}
assert build_deps["descendants"] == {"lint", "test", "deploy"}

# ============================================================================
# 6. Critical Path (Longest Path) in DAGs (Unweighted & Weighted)
# ============================================================================
# Unweighted DAG longest path
longest_path, longest_len = dag_ops.compute_critical_path(dag)
assert longest_len == 3.0
assert longest_path[0] == "fetch"
assert longest_path[-1] == "deploy"
assert len(longest_path) == 4

# Weighted DAG critical path
stage_edges = [
    ("start", "step1"),
    ("step1", "end"),
    ("start", "heavy_step"),
    ("heavy_step", "end"),
]
stage_weights = {
    ("start", "step1"): 1.5,
    ("step1", "end"): 2.0,
    ("start", "heavy_step"): 5.0,
    ("heavy_step", "end"): 8.5,
}
weighted_dag = dag_ops.build_dependency_graph(stage_edges, stage_weights)
crit_path, crit_len = dag_ops.compute_critical_path(weighted_dag, weight="weight")
assert crit_path == ["start", "heavy_step", "end"]
assert crit_len == 13.5

# ============================================================================
# 7. Transitive Reduction Prunes Redundant Transitive Edges
# ============================================================================
# Add shortcut edge directly from "fetch" to "deploy"
dag.add_edge("fetch", "deploy")
assert ("fetch", "deploy") in dag.edges()

reduced = dag_ops.simplify_transitive_dependencies(dag)
assert ("fetch", "deploy") not in reduced.edges()
assert set(reduced.nodes()) == set(dag.nodes())
assert nx.has_path(reduced, "fetch", "deploy") is True

print(f"CONTRACT PASS: networkx {nx.__version__} DAG analysis, topological sorting, cycle detection, and lineage verified.")