refactor: land completed PLF-135 through PLF-146 changes

src/tagi/analyzer/dependency_graph.py

@@ -100,14 +100,14 @@ def levels(self) -> List[List[str]]:
         Returns:
             List of levels, each containing files that can be committed together
         """
-        result: List[List[str]] = []
+        commit_groups: List[List[str]] = []
 
         while self.queue:
             level = self._drain_level()
             if level:
-                result.append(level)
+                commit_groups.append(level)
 
-        return result
+        return commit_groups
 
     def _drain_level(self) -> List[str]:
         """Pop one queue level and enqueue dependents that become unblocked.

src/tagi/cli/provider_commands.py

@@ -26,12 +26,11 @@ def detect_provider_command(repo_path: str = ".") -> Optional[str]:
 
 def _current_branch(repo_path: str) -> str:
     """Resolve the current git branch for the repository."""
-    import subprocess
-    result = subprocess.run(
+    from tagi.utils.commands import run_command
+    return run_command(
         ["git", "rev-parse", "--abbrev-ref", "HEAD"],
-        cwd=repo_path, capture_output=True, text=True, check=False,
-    )
-    return result.stdout.strip() or "main"
+        repo_path,
+    ).stdout.strip() or "main"
 
 
 def _pr_spec(spec: PrSpec, repo_path: str) -> PrSpec:

src/tagi/composer/_tags.py

@@ -19,11 +19,11 @@ def all_tag_values(changes: List[Change]) -> List[str]:
 
 def summary_tag(changes: List[Change]) -> str:
     """Choose a stable tag prefix for a commit spanning one or more changes."""
-    tags = all_tags(changes)
-    if not tags:
+    unique_tags = all_tags(changes)
+    if not unique_tags:
         return "#small"
-    if len(tags) == 1:
-        return tags[0].value
+    if len(unique_tags) == 1:
+        return unique_tags[0].value
     return "#all"
 
 

src/tagi/composer/formats.py

@@ -45,8 +45,8 @@ def generate_conventional_message(changes: List[Change]) -> str:
         return "chore: empty commit"
 
     # Determine type from tags
-    tags = set(all_tags(changes))
-    commit_type = _commit_type(tags)
+    tag_set = set(all_tags(changes))
+    commit_type = _commit_type(tag_set)
 
     # Determine scope from file paths
     scope = infer_scope(changes)
@@ -55,7 +55,7 @@ def generate_conventional_message(changes: List[Change]) -> str:
     description = _describe_changes(changes)
 
     # Add optional breaking change indicator
-    breaking = "!" if Tag.RISKY in tags else ""
+    breaking = "!" if Tag.RISKY in tag_set else ""
 
     if scope:
         return f"{commit_type}({scope}){breaking}: {description}"

src/tagi/config.py

@@ -4,6 +4,8 @@
 from pathlib import Path
 from typing import Dict, List, Optional
 
+from tagi.utils.paths import path_matches
+
 try:
     import tomli
 except ImportError:
@@ -70,12 +72,10 @@ def _apply_sections(self, data):
 
     def get_tag_for_path(self, path: str) -> Optional[str]:
         """Get custom tag for a file path based on rules."""
-        path_lower = path.lower()
-
         for pattern, tag in self.custom_rules.items():
-            if pattern.lower() in path_lower:
+            if path_matches(path, pattern):
                 return tag
-        
+
         return None
 
     def get_custom_tags_for_pattern(self, pattern: str) -> List[str]:
@@ -88,14 +88,12 @@ def get_tag_color(self, tag: str) -> Optional[str]:
 
     def get_heuristics_for_path(self, path: str) -> List[str]:
         """Get custom heuristic tags for a file path."""
-        path_lower = path.lower()
-        tags = []
-
-        for pattern, pattern_tags in self.custom_heuristics.items():
-            if pattern.lower() in path_lower:
-                tags.extend(pattern_tags)
-
-        return tags
+        return [
+            pattern_tag
+            for pattern, pattern_tags in self.custom_heuristics.items()
+            if path_matches(path, pattern)
+            for pattern_tag in pattern_tags
+        ]
 
     def get_tags_for_path(self, path: str) -> List[str]:
         """Get all custom tags for a file path: rule tag first, then heuristic tags."""
@@ -112,13 +110,7 @@ def get_template(self, template_name: str) -> Optional[str]:
 
     def should_ignore(self, path: str) -> bool:
         """Check if a path should be ignored based on ignore patterns."""
-        path_lower = path.lower()
-
-        for pattern in self.ignore_patterns:
-            if pattern.lower() in path_lower:
-                return True
-
-        return False
+        return any(path_matches(path, pattern) for pattern in self.ignore_patterns)
 
 
 def load_config(repo_path: str = ".") -> Config:

src/tagi/executor/git.py

@@ -1,9 +1,10 @@
 """Git executor module for running git commands."""
 
-import subprocess
 from subprocess import CompletedProcess
 from typing import List, Optional
 
+from tagi.utils.commands import run_command
+
 
 class GitExecutor:
     """Executor for git commands."""
@@ -13,40 +14,34 @@ def __init__(self, repo_path: str = "."):
 
     def _run_command(self, command: list[str]) -> CompletedProcess:
         """Run a command in the repository and return the result."""
-        return subprocess.run(
-            command,
-            cwd=self.repo_path,
-            capture_output=True,
-            text=True,
-            check=False
-        )
+        return run_command(command, self.repo_path)
 
     def add(self, files: List[str]) -> bool:
         """Stage files for commit."""
         if not files:
             return False
-        result = self._run_command(["git", "add", *files])
-        if result.returncode != 0:
-            raise RuntimeError(f"Failed to stage files: {result.stderr}")
+        staged = self._run_command(["git", "add", *files])
+        if staged.returncode != 0:
+            raise RuntimeError(f"Failed to stage files: {staged.stderr}")
         return True
 
     def commit(self, message: str, allow_empty: bool = False) -> bool:
         """Commit staged changes."""
-        result = self._run_command(
+        committed = self._run_command(
             ["git", "commit", "-m", message] + (["--allow-empty"] if allow_empty else [])
         )
-        if result.returncode != 0:
-            raise RuntimeError(f"Failed to commit: {result.stderr}")
+        if committed.returncode != 0:
+            raise RuntimeError(f"Failed to commit: {committed.stderr}")
         return True
 
     def push(self, remote: str = "origin", branch: Optional[str] = None, force: bool = False) -> bool:
         """Push commits to remote."""
-        result = self._run_command(
+        pushed = self._run_command(
             (["git", "push", remote, branch] if branch else ["git", "push"])
             + (["--force"] if force else [])
         )
-        if result.returncode != 0:
-            raise RuntimeError(f"Failed to push: {result.stderr}")
+        if pushed.returncode != 0:
+            raise RuntimeError(f"Failed to push: {pushed.stderr}")
         return True
 
     def status(self) -> str:
@@ -55,16 +50,16 @@ def status(self) -> str:
 
     def get_current_branch(self) -> str:
         """Get the current branch name."""
-        result = self._run_command(["git", "branch", "--show-current"])
-        if result.returncode == 0:
-            return result.stdout.strip()
+        branch = self._run_command(["git", "branch", "--show-current"])
+        if branch.returncode == 0:
+            return branch.stdout.strip()
         return "main"
 
     def get_remote_url(self, remote: str = "origin") -> Optional[str]:
         """Get the remote URL."""
-        result = self._run_command(["git", "remote", "get-url", remote])
-        if result.returncode == 0:
-            return result.stdout.strip()
+        url = self._run_command(["git", "remote", "get-url", remote])
+        if url.returncode == 0:
+            return url.stdout.strip()
         return None
 
     def has_staged_changes(self) -> bool:

src/tagi/heuristics/metrics.py

@@ -3,6 +3,7 @@
 import math
 from typing import List
 from tagi.models.change import Change, ChangeMetrics, ChangeType
+from tagi.utils.paths import path_endswith, path_matches
 
 
 _COMPLEXITY_EXTENSION_WEIGHTS = (
@@ -32,18 +33,16 @@
 
 def _extension_complexity_weight(path: str) -> float:
     """Return the complexity weight for a file path (first matching suffix wins)."""
-    path_lower = path.lower()
     for suffixes, weight in _COMPLEXITY_EXTENSION_WEIGHTS:
-        if path_lower.endswith(suffixes):
+        if path_endswith(path, suffixes):
             return weight
     return _DEFAULT_COMPLEXITY_EXTENSION_WEIGHT
 
 
 def _extension_impact_weight(path: str) -> float:
     """Return the impact weight for a file path (first matching suffix wins)."""
-    path_lower = path.lower()
     for suffixes, weight in _IMPACT_EXTENSION_WEIGHTS:
-        if path_lower.endswith(suffixes):
+        if path_endswith(path, suffixes):
             return weight
     return _DEFAULT_IMPACT_EXTENSION_WEIGHT
 
@@ -134,14 +133,12 @@ def _calculate_stability(change: Change) -> float:
 
 def _calculate_test_impact(change: Change) -> float:
     """Calculate test coverage impact (0-1)."""
-    path_lower = change.path.lower()
-
     # Test files have high impact on test coverage
-    if 'test' in path_lower or 'spec' in path_lower:
+    if path_matches(change.path, 'test') or path_matches(change.path, 'spec'):
         return 0.9
 
     # Source files have moderate impact
-    if path_lower.endswith(('.py', '.js', '.ts')):
+    if path_endswith(change.path, ('.py', '.js', '.ts')):
         return 0.6
 
     # Other files have low impact
@@ -151,14 +148,12 @@ def _calculate_test_impact(change: Change) -> float:
 def _calculate_dependency_depth(change: Change) -> int:
     """Calculate dependency depth (simplified)."""
     # This is a simplified version - real implementation would parse imports
-    path_lower = change.path.lower()
-
-    if path_lower.endswith('.py'):
+    if path_endswith(change.path, ('.py',)):
         # Python files typically have more dependencies
         return 3
-    elif path_lower.endswith(('.js', '.ts', '.jsx', '.tsx')):
+    elif path_endswith(change.path, ('.js', '.ts', '.jsx', '.tsx')):
         return 2
-    elif path_lower.endswith(('.json', '.yaml', '.yml')):
+    elif path_endswith(change.path, ('.json', '.yaml', '.yml')):
         return 1
     else:
         return 0

src/tagi/heuristics/tags.py

@@ -5,6 +5,7 @@
 from tagi.config import load_config
 from tagi.models import Change, ChangeType, Tag
 from tagi.scanner.files import count_lines_changed
+from tagi.utils.paths import path_matches
 from .scoring import calculate_risk_score
 from .metrics import calculate_metrics
 
@@ -47,13 +48,13 @@ def _tag_change(change: Change, repo_path: str, custom_tags_for) -> None:
 
 def _custom_config_tags(custom_tags_for, path: str):
     """Convert configured custom tag names into valid Tag values."""
-    tags = []
+    valid_tags = []
     for custom_tag in custom_tags_for(path):
         try:
-            tags.append(Tag(custom_tag))
+            valid_tags.append(Tag(custom_tag))
         except ValueError:
             pass  # Invalid tag, skip
-    return tags
+    return valid_tags
 
 
 def _size_tags(lines_changed: int, has_other_tags: bool):
@@ -68,8 +69,6 @@ def _size_tags(lines_changed: int, has_other_tags: bool):
 
 def apply_path_tags(change: Change, lines_changed: int) -> List[Tag]:
     """Apply path-based heuristic tags to a change."""
-    path_lower = change.path.lower()
-
     # Pattern mapping for tag detection
     tag_patterns = [
         (['requirements', 'package.json', 'poetry.lock', 'pyproject.toml', 'cargo.toml', 'go.mod', 'yarn.lock', 'pnpm-lock.yaml', 'package-lock.json', 'gemfile', 'composer.json'], Tag.DEPS),
@@ -85,5 +84,5 @@ def apply_path_tags(change: Change, lines_changed: int) -> List[Tag]:
     return [
         tag
         for patterns, tag in tag_patterns
-        if any(pattern in path_lower for pattern in patterns)
+        if any(path_matches(change.path, pattern) for pattern in patterns)
     ]

src/tagi/planner/branch_grouper.py

@@ -15,26 +15,24 @@ def group_by_branch(changes: List[Change], repo_path: str = ".") -> Dict[str, Li
         Dictionary mapping branch names to lists of changes
     """
     from tagi.executor.git import GitExecutor
-    import subprocess
-    
+    from tagi.utils.commands import run_command
+
     executor = GitExecutor(repo_path)
     current_branch = executor.get_current_branch()
-    
+
     # Get branch history for each file
     branch_groups: Dict[str, List[Change]] = {}
-    
+
     for change in changes:
         try:
             # Get the branch where the file was last modified
-            result = subprocess.run(
+            contains = run_command(
                 ["git", "branch", "--contains", "HEAD", "--", change.path],
-                cwd=repo_path,
-                capture_output=True,
-                text=True
+                repo_path,
             )
-            
-            if result.returncode == 0:
-                branches = result.stdout.strip().split('\n')
+
+            if contains.returncode == 0:
+                branches = contains.stdout.strip().split('\n')
                 # Clean up branch names (remove * prefix)
                 branches = [b.strip().replace('*', '').strip() for b in branches if b.strip()]
 
@@ -64,21 +62,16 @@ def get_branch_info(repo_path: str = ".") -> Dict[str, str]:
     Returns:
         Dictionary mapping branch names to their latest commit hashes
     """
-    import subprocess
-    
+    from tagi.utils.commands import run_command
+
     try:
-        result = subprocess.run(
-            ["git", "branch", "-a"],
-            cwd=repo_path,
-            capture_output=True,
-            text=True
-        )
-
-        if result.returncode != 0:
+        branch_listing = run_command(["git", "branch", "-a"], repo_path)
+
+        if branch_listing.returncode != 0:
             return {}
-        
+
         branches = {}
-        for line in result.stdout.strip().split('\n'):
+        for line in branch_listing.stdout.strip().split('\n'):
             branch = line.strip().replace('*', '').strip()
             if branch:
                 branches[branch] = branch  # Could be extended to include commit hash