abahgat · January 26, 2026 19:57
diff --git a/dependency_graph.py b/dependency_graph.py
 #!/usr/bin/env python3
 """
 Dependency Graph Analyzer for Linear Issues

 This script helps visualize and validate dependency relationships between Linear issues
 for stacked PR workflows.

 Usage:
    python dependency_graph.py --help
    python dependency_graph.py --visualize <issue_data.json>
    python dependency_graph.py --check-cycles <issue_data.json>
    python dependency_graph.py --work-order <issue_data.json>
 """

 import json
 import sys
 import argparse
 from collections import defaultdict, deque
 from typing import Dict, List, Set, Tuple, Optional


 class Issue:
    """Represents a Linear issue with its dependencies"""

    def __init__(self, data: dict):
        self.id = data.get("id")
        self.identifier = data.get("identifier", "UNKNOWN")
        self.title = data.get("title", "Untitled")
        self.priority = data.get("priority", 0)
        self.parent_id = (
            data.get("parent", {}).get("id") if data.get("parent") else None
        )

        # Extract relationships
        relations = data.get("relations", {})
        self.blocked_by = [r.get("id") for r in relations.get("blockedByIssues", [])]
        self.blocks = [r.get("id") for r in relations.get("blockingIssues", [])]
        self.related = [r.get("id") for r in relations.get("relatedIssues", [])]

    def __repr__(self):
        return f"Issue({self.identifier}: {self.title})"


 class DependencyGraph:
    """Builds and analyzes dependency graphs for Linear issues"""

    def __init__(self, issues: List[Issue]):
        self.issues = {issue.id: issue for issue in issues}
        self.graph = self._build_graph()

    def _build_graph(self) -> Dict[str, Dict]:
        """Build adjacency list representation of dependency graph"""
        graph = {}

        # Initialize nodes
        for issue_id, issue in self.issues.items():
            graph[issue_id] = {
                "issue": issue,
                "dependencies": set(),  # Issues this depends on (must complete first)
                "dependents": set(),  # Issues that depend on this
            }

        # Add blocked-by relationships
        for issue_id, issue in self.issues.items():
            for blocker_id in issue.blocked_by:
                if blocker_id in graph:
                    graph[issue_id]["dependencies"].add(blocker_id)
                    graph[blocker_id]["dependents"].add(issue_id)

        # Add parent-child relationships (if not already covered)
        for issue_id, issue in self.issues.items():
            if issue.parent_id and issue.parent_id in graph:
                # Check if parent blocks child or child blocks parent
                parent_blocks_child = issue.parent_id in issue.blocked_by
                child_blocks_parent = (
                    issue_id in self.issues[issue.parent_id].blocked_by
                )

                if not parent_blocks_child and not child_blocks_parent:
                    # Default: parent comes before child
                    graph[issue_id]["dependencies"].add(issue.parent_id)
                    graph[issue.parent_id]["dependents"].add(issue_id)

        return graph

    def topological_sort(self) -> Tuple[List[str], bool]:
        """
        Perform topological sort to determine valid work order.

        Returns:
            (ordered_ids, success): List of issue IDs in work order, and whether sort succeeded
        """
        in_degree = {node: len(self.graph[node]["dependencies"]) for node in self.graph}
        queue = deque([node for node, degree in in_degree.items() if degree == 0])
        result = []

        while queue:
            # Sort by priority (higher priority first) for stable ordering
            queue = deque(sorted(queue, key=lambda n: -self.graph[n]["issue"].priority))
            node = queue.popleft()
            result.append(node)

            # Reduce in-degree for dependents
            for dependent in self.graph[node]["dependents"]:
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    queue.append(dependent)

        # Check if all nodes were visited (no cycles)
        success = len(result) == len(self.graph)
        return result, success

    def find_cycles(self) -> List[List[str]]:
        """
        Detect cycles in the dependency graph using DFS.

        Returns:
            List of cycles, where each cycle is a list of issue IDs
        """
        visited = set()
        rec_stack = set()
        cycles = []

        def dfs(node, path):
            visited.add(node)
            rec_stack.add(node)
            path.append(node)

            for dependent in self.graph[node]["dependents"]:
                if dependent not in visited:
                    dfs(dependent, path[:])
                elif dependent in rec_stack:
                    # Found a cycle
                    cycle_start = path.index(dependent)
                    cycle = path[cycle_start:] + [dependent]
                    cycles.append(cycle)

            rec_stack.remove(node)

        for node in self.graph:
            if node not in visited:
                dfs(node, [])

        return cycles

    def get_roots(self) -> List[str]:
        """Get issues with no dependencies (can start immediately)"""
        return [
            issue_id
            for issue_id, data in self.graph.items()
            if len(data["dependencies"]) == 0
        ]

    def get_leaves(self) -> List[str]:
        """Get issues with no dependents (end of chains)"""
        return [
            issue_id
            for issue_id, data in self.graph.items()
            if len(data["dependents"]) == 0
        ]

    def visualize_ascii(self) -> str:
        """Generate ASCII art visualization of the dependency graph"""
        order, success = self.topological_sort()

        if not success:
            cycles = self.find_cycles()
            output = ["⚠️  DEPENDENCY CYCLE DETECTED!", ""]
            for cycle in cycles:
                cycle_strs = [self.issues[id].identifier for id in cycle]
                output.append(f"Cycle: {' → '.join(cycle_strs)}")
            output.append("")
            return "\n".join(output)

        output = ["Dependency Graph (Work Order):", ""]

        for i, issue_id in enumerate(order, 1):
            issue = self.graph[issue_id]["issue"]
            deps = self.graph[issue_id]["dependencies"]

            prefix = f"{i}. "
            output.append(f"{prefix}{issue.identifier} - {issue.title}")

            if deps:
                dep_strs = [self.issues[d].identifier for d in deps]
                output.append(f"   └─ Depends on: {', '.join(dep_strs)}")

            output.append("")

        return "\n".join(output)

    def to_mermaid(self) -> str:
        """Generate Mermaid diagram syntax for the dependency graph"""
        lines = ["graph TD"]

        # Add nodes
        for issue_id, issue in self.issues.items():
            label = f"{issue.identifier}: {issue.title[:30]}"
            lines.append(f'    {issue_id}["{label}"]')

        # Add edges
        for issue_id, data in self.graph.items():
            for dep_id in data["dependencies"]:
                lines.append(f"    {dep_id} --> {issue_id}")

        return "\n".join(lines)

    def get_statistics(self) -> Dict:
        """Get statistics about the dependency graph"""
        order, has_cycle = self.topological_sort()

        return {
            "total_issues": len(self.issues),
            "has_cycle": not has_cycle,
            "num_roots": len(self.get_roots()),
            "num_leaves": len(self.get_leaves()),
            "max_depth": self._calculate_max_depth(),
            "avg_dependencies": sum(len(d["dependencies"]) for d in self.graph.values())
            / len(self.graph),
        }

    def _calculate_max_depth(self) -> int:
        """Calculate the maximum depth of the dependency tree"""

        def dfs_depth(node, visited):
            if node in visited:
                return 0
            visited.add(node)
            max_child_depth = 0
            for dependent in self.graph[node]["dependents"]:
                max_child_depth = max(max_child_depth, dfs_depth(dependent, visited))
            return 1 + max_child_depth

        roots = self.get_roots()
        return max(dfs_depth(root, set()) for root in roots) if roots else 0


 def load_issues_from_json(filepath: str) -> List[Issue]:
    """Load issues from a JSON file"""
    with open(filepath, "r") as f:
        data = json.load(f)

    # Support both single issue and array of issues
    if isinstance(data, list):
        return [Issue(item) for item in data]
    else:
        return [Issue(data)]


 def main():
    parser = argparse.ArgumentParser(
        description="Analyze dependency graph for Linear issues",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    parser.add_argument("input_file", help="JSON file containing Linear issue data")

    parser.add_argument(
        "--visualize",
        "-v",
        action="store_true",
        help="Display ASCII visualization of the graph",
    )

    parser.add_argument(
        "--mermaid", "-m", action="store_true", help="Output Mermaid diagram syntax"
    )

    parser.add_argument(
        "--check-cycles", "-c", action="store_true", help="Check for dependency cycles"
    )

    parser.add_argument(
        "--work-order",
        "-w",
        action="store_true",
        help="Output work order (topological sort)",
    )

    parser.add_argument(
        "--stats", "-s", action="store_true", help="Display graph statistics"
    )

    args = parser.parse_args()

    # Load issues
    try:
        issues = load_issues_from_json(args.input_file)
    except FileNotFoundError:
        print(f"Error: File '{args.input_file}' not found", file=sys.stderr)
        return 1
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in '{args.input_file}': {e}", file=sys.stderr)
        return 1

    if not issues:
        print("Error: No issues found in input file", file=sys.stderr)
        return 1

    # Build graph
    graph = DependencyGraph(issues)

    # If no flags specified, show visualization by default
    if not any(
        [args.visualize, args.mermaid, args.check_cycles, args.work_order, args.stats]
    ):
        args.visualize = True

    # Execute requested operations
    if args.visualize:
        print(graph.visualize_ascii())

    if args.mermaid:
        print(graph.to_mermaid())

    if args.check_cycles:
        cycles = graph.find_cycles()
        if cycles:
            print("⚠️  Dependency cycles detected!")
            for i, cycle in enumerate(cycles, 1):
                cycle_strs = [graph.issues[id].identifier for id in cycle]
                print(f"\nCycle {i}: {' → '.join(cycle_strs)}")
            return 1
        else:
            print("✓ No dependency cycles detected")

    if args.work_order:
        order, success = graph.topological_sort()
        if success:
            print("Work Order:")
            for i, issue_id in enumerate(order, 1):
                issue = graph.issues[issue_id]
                print(f"{i}. {issue.identifier} - {issue.title}")
        else:
            print(
                "Error: Cannot determine work order (dependency cycle exists)",
                file=sys.stderr,
            )
            return 1

    if args.stats:
        stats = graph.get_statistics()
        print("Graph Statistics:")
        print(f"  Total issues: {stats['total_issues']}")
        print(f"  Has cycles: {stats['has_cycle']}")
        print(f"  Root issues (no dependencies): {stats['num_roots']}")
        print(f"  Leaf issues (no dependents): {stats['num_leaves']}")
        print(f"  Maximum depth: {stats['max_depth']}")
        print(f"  Average dependencies per issue: {stats['avg_dependencies']:.2f}")

    return 0


 if __name__ == "__main__":
    sys.exit(main())
diff --git a/REFERENCE.md b/REFERENCE.md
diff --git a/SKILL.md b/SKILL.md
diff --git a/WORKFLOW.md b/WORKFLOW.md
	#!/usr/bin/env python3
	"""
	Dependency Graph Analyzer for Linear Issues

	This script helps visualize and validate dependency relationships between Linear issues
	for stacked PR workflows.

	Usage:
	python dependency_graph.py --help
	python dependency_graph.py --visualize <issue_data.json>
	python dependency_graph.py --check-cycles <issue_data.json>
	python dependency_graph.py --work-order <issue_data.json>
	"""

	import json
	import sys
	import argparse
	from collections import defaultdict, deque
	from typing import Dict, List, Set, Tuple, Optional


	class Issue:
	"""Represents a Linear issue with its dependencies"""

	def __init__(self, data: dict):
	self.id = data.get("id")
	self.identifier = data.get("identifier", "UNKNOWN")
	self.title = data.get("title", "Untitled")
	self.priority = data.get("priority", 0)
	self.parent_id = (
	data.get("parent", {}).get("id") if data.get("parent") else None
	)

	# Extract relationships
	relations = data.get("relations", {})
	self.blocked_by = [r.get("id") for r in relations.get("blockedByIssues", [])]
	self.blocks = [r.get("id") for r in relations.get("blockingIssues", [])]
	self.related = [r.get("id") for r in relations.get("relatedIssues", [])]

	def __repr__(self):
	return f"Issue({self.identifier}: {self.title})"


	class DependencyGraph:
	"""Builds and analyzes dependency graphs for Linear issues"""

	def __init__(self, issues: List[Issue]):
	self.issues = {issue.id: issue for issue in issues}
	self.graph = self._build_graph()

	def _build_graph(self) -> Dict[str, Dict]:
	"""Build adjacency list representation of dependency graph"""
	graph = {}

	# Initialize nodes
	for issue_id, issue in self.issues.items():
	graph[issue_id] = {
	"issue": issue,
	"dependencies": set(), # Issues this depends on (must complete first)
	"dependents": set(), # Issues that depend on this
	}

	# Add blocked-by relationships
	for issue_id, issue in self.issues.items():
	for blocker_id in issue.blocked_by:
	if blocker_id in graph:
	graph[issue_id]["dependencies"].add(blocker_id)
	graph[blocker_id]["dependents"].add(issue_id)

	# Add parent-child relationships (if not already covered)
	for issue_id, issue in self.issues.items():
	if issue.parent_id and issue.parent_id in graph:
	# Check if parent blocks child or child blocks parent
	parent_blocks_child = issue.parent_id in issue.blocked_by
	child_blocks_parent = (
	issue_id in self.issues[issue.parent_id].blocked_by
	)

	if not parent_blocks_child and not child_blocks_parent:
	# Default: parent comes before child
	graph[issue_id]["dependencies"].add(issue.parent_id)
	graph[issue.parent_id]["dependents"].add(issue_id)

	return graph

	def topological_sort(self) -> Tuple[List[str], bool]:
	"""
	Perform topological sort to determine valid work order.

	Returns:
	(ordered_ids, success): List of issue IDs in work order, and whether sort succeeded
	"""
	in_degree = {node: len(self.graph[node]["dependencies"]) for node in self.graph}
	queue = deque([node for node, degree in in_degree.items() if degree == 0])
	result = []

	while queue:
	# Sort by priority (higher priority first) for stable ordering
	queue = deque(sorted(queue, key=lambda n: -self.graph[n]["issue"].priority))
	node = queue.popleft()
	result.append(node)

	# Reduce in-degree for dependents
	for dependent in self.graph[node]["dependents"]:
	in_degree[dependent] -= 1
	if in_degree[dependent] == 0:
	queue.append(dependent)

	# Check if all nodes were visited (no cycles)
	success = len(result) == len(self.graph)
	return result, success

	def find_cycles(self) -> List[List[str]]:
	"""
	Detect cycles in the dependency graph using DFS.

	Returns:
	List of cycles, where each cycle is a list of issue IDs
	"""
	visited = set()
	rec_stack = set()
	cycles = []

	def dfs(node, path):
	visited.add(node)
	rec_stack.add(node)
	path.append(node)

	for dependent in self.graph[node]["dependents"]:
	if dependent not in visited:
	dfs(dependent, path[:])
	elif dependent in rec_stack:
	# Found a cycle
	cycle_start = path.index(dependent)
	cycle = path[cycle_start:] + [dependent]
	cycles.append(cycle)

	rec_stack.remove(node)

	for node in self.graph:
	if node not in visited:
	dfs(node, [])

	return cycles

	def get_roots(self) -> List[str]:
	"""Get issues with no dependencies (can start immediately)"""
	return [
	issue_id
	for issue_id, data in self.graph.items()
	if len(data["dependencies"]) == 0
	]

	def get_leaves(self) -> List[str]:
	"""Get issues with no dependents (end of chains)"""
	return [
	issue_id
	for issue_id, data in self.graph.items()
	if len(data["dependents"]) == 0
	]

	def visualize_ascii(self) -> str:
	"""Generate ASCII art visualization of the dependency graph"""
	order, success = self.topological_sort()

	if not success:
	cycles = self.find_cycles()
	output = ["⚠️ DEPENDENCY CYCLE DETECTED!", ""]
	for cycle in cycles:
	cycle_strs = [self.issues[id].identifier for id in cycle]
	output.append(f"Cycle: {' → '.join(cycle_strs)}")
	output.append("")
	return "\n".join(output)

	output = ["Dependency Graph (Work Order):", ""]

	for i, issue_id in enumerate(order, 1):
	issue = self.graph[issue_id]["issue"]
	deps = self.graph[issue_id]["dependencies"]

	prefix = f"{i}. "
	output.append(f"{prefix}{issue.identifier} - {issue.title}")

	if deps:
	dep_strs = [self.issues[d].identifier for d in deps]
	output.append(f" └─ Depends on: {', '.join(dep_strs)}")

	output.append("")

	return "\n".join(output)

	def to_mermaid(self) -> str:
	"""Generate Mermaid diagram syntax for the dependency graph"""
	lines = ["graph TD"]

	# Add nodes
	for issue_id, issue in self.issues.items():
	label = f"{issue.identifier}: {issue.title[:30]}"
	lines.append(f' {issue_id}["{label}"]')

	# Add edges
	for issue_id, data in self.graph.items():
	for dep_id in data["dependencies"]:
	lines.append(f" {dep_id} --> {issue_id}")

	return "\n".join(lines)

	def get_statistics(self) -> Dict:
	"""Get statistics about the dependency graph"""
	order, has_cycle = self.topological_sort()

	return {
	"total_issues": len(self.issues),
	"has_cycle": not has_cycle,
	"num_roots": len(self.get_roots()),
	"num_leaves": len(self.get_leaves()),
	"max_depth": self._calculate_max_depth(),
	"avg_dependencies": sum(len(d["dependencies"]) for d in self.graph.values())
	/ len(self.graph),
	}

	def _calculate_max_depth(self) -> int:
	"""Calculate the maximum depth of the dependency tree"""

	def dfs_depth(node, visited):
	if node in visited:
	return 0
	visited.add(node)
	max_child_depth = 0
	for dependent in self.graph[node]["dependents"]:
	max_child_depth = max(max_child_depth, dfs_depth(dependent, visited))
	return 1 + max_child_depth

	roots = self.get_roots()
	return max(dfs_depth(root, set()) for root in roots) if roots else 0


	def load_issues_from_json(filepath: str) -> List[Issue]:
	"""Load issues from a JSON file"""
	with open(filepath, "r") as f:
	data = json.load(f)

	# Support both single issue and array of issues
	if isinstance(data, list):
	return [Issue(item) for item in data]
	else:
	return [Issue(data)]


	def main():
	parser = argparse.ArgumentParser(
	description="Analyze dependency graph for Linear issues",
	formatter_class=argparse.RawDescriptionHelpFormatter,
	)

	parser.add_argument("input_file", help="JSON file containing Linear issue data")

	parser.add_argument(
	"--visualize",
	"-v",
	action="store_true",
	help="Display ASCII visualization of the graph",
	)

	parser.add_argument(
	"--mermaid", "-m", action="store_true", help="Output Mermaid diagram syntax"
	)

	parser.add_argument(
	"--check-cycles", "-c", action="store_true", help="Check for dependency cycles"
	)

	parser.add_argument(
	"--work-order",
	"-w",
	action="store_true",
	help="Output work order (topological sort)",
	)

	parser.add_argument(
	"--stats", "-s", action="store_true", help="Display graph statistics"
	)

	args = parser.parse_args()

	# Load issues
	try:
	issues = load_issues_from_json(args.input_file)
	except FileNotFoundError:
	print(f"Error: File '{args.input_file}' not found", file=sys.stderr)
	return 1
	except json.JSONDecodeError as e:
	print(f"Error: Invalid JSON in '{args.input_file}': {e}", file=sys.stderr)
	return 1

	if not issues:
	print("Error: No issues found in input file", file=sys.stderr)
	return 1

	# Build graph
	graph = DependencyGraph(issues)

	# If no flags specified, show visualization by default
	if not any(
	[args.visualize, args.mermaid, args.check_cycles, args.work_order, args.stats]
	):
	args.visualize = True

	# Execute requested operations
	if args.visualize:
	print(graph.visualize_ascii())

	if args.mermaid:
	print(graph.to_mermaid())

	if args.check_cycles:
	cycles = graph.find_cycles()
	if cycles:
	print("⚠️ Dependency cycles detected!")
	for i, cycle in enumerate(cycles, 1):
	cycle_strs = [graph.issues[id].identifier for id in cycle]
	print(f"\nCycle {i}: {' → '.join(cycle_strs)}")
	return 1
	else:
	print("✓ No dependency cycles detected")

	if args.work_order:
	order, success = graph.topological_sort()
	if success:
	print("Work Order:")
	for i, issue_id in enumerate(order, 1):
	issue = graph.issues[issue_id]
	print(f"{i}. {issue.identifier} - {issue.title}")
	else:
	print(
	"Error: Cannot determine work order (dependency cycle exists)",
	file=sys.stderr,
	)
	return 1

	if args.stats:
	stats = graph.get_statistics()
	print("Graph Statistics:")
	print(f" Total issues: {stats['total_issues']}")
	print(f" Has cycles: {stats['has_cycle']}")
	print(f" Root issues (no dependencies): {stats['num_roots']}")
	print(f" Leaf issues (no dependents): {stats['num_leaves']}")
	print(f" Maximum depth: {stats['max_depth']}")
	print(f" Average dependencies per issue: {stats['avg_dependencies']:.2f}")

	return 0


	if __name__ == "__main__":
	sys.exit(main())
No results found