refactor: split oversized test and compression modules

Decompose several oversized test suites and compression strategy logic into smaller modules while preserving import compatibility and behavior, improving maintainability without call-site changes. Harden duplicate test filename meta-check to evaluate tracked tests so concurrent untracked work does not produce false local failures.

Made-with: Cursor

Author: Mateusz

dev/scripts/analyze_time_violations.py

@@ -6,12 +6,13 @@
 import sys
 from collections import defaultdict
 from pathlib import Path
+from typing import Any
 
 # Add project root to path
 repo_root = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(repo_root))
 
-from tests.unit.test_time_usage_linter import _scan_repo_for_time_usage
+from tests.unit.support.time_usage_linter_scanner import scan_repo_for_time_usage
 from tests.utils.time_policy import load_allowlist
 
 
@@ -24,10 +25,10 @@ def main() -> None:
         cache_path.unlink()
 
     allowlist = load_allowlist()
-    findings = _scan_repo_for_time_usage(repo_root, allowlist)
+    findings = scan_repo_for_time_usage(repo_root, allowlist)
 
     # Categorize by file and type
-    by_file: dict[str, list[dict[str, any]]] = defaultdict(list)
+    by_file: dict[str, list[dict[str, Any]]] = defaultdict(list)
     by_type: dict[str, int] = defaultdict(int)
 
     for finding in findings:

dev/scripts/create_time_usage_inventory.py

@@ -10,14 +10,15 @@
 import sys
 from collections import defaultdict
 from pathlib import Path
+from typing import Any
 
 # Add repo root to path
 repo_root = Path(__file__).parent.parent.parent
 sys.path.insert(0, str(repo_root))
 
-from tests.unit.test_time_usage_linter import (
+from tests.unit.support.time_usage_linter_scanner import (
     LintFinding,
-    _scan_repo_for_time_usage,
+    scan_repo_for_time_usage,
 )
 from tests.utils.time_policy import load_allowlist
 
@@ -99,7 +100,7 @@ def create_inventory(repo_root: Path, output_path: Path) -> None:
     """
     print("Scanning repository for time usage violations...")
     allowlist = load_allowlist()
-    findings = _scan_repo_for_time_usage(repo_root, allowlist)
+    findings = scan_repo_for_time_usage(repo_root, allowlist)
 
     print(f"Found {len(findings)} violations")
 
@@ -143,7 +144,7 @@ def create_inventory(repo_root: Path, output_path: Path) -> None:
         classified[classification].append(entry)
 
     # Create summary statistics
-    summary = {
+    summary: dict[str, Any] = {
         "total_violations": len(findings),
         "by_rule": {
             rule: len(findings_list) for rule, findings_list in by_rule.items()

dev/scripts/pre-commit-hook.py

@@ -357,7 +357,7 @@ def run_stall_linter_on_staged_tests() -> bool:
         "pytest",
         "-n",
         "0",
-        "tests/unit/test_stall_linter.py",
+        "tests/unit/stall_linter/",
     ]
     for file_path in target_test_files:
         pytest_args.extend(["--stall-lint-file", file_path])

src/core/services/_compression_strategies_common.py

@@ -0,0 +1,78 @@
+"""Shared regexes, logging, and helpers for compression strategies."""
+
+from __future__ import annotations
+
+import logging
+import re
+
+_CSI_RE = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]")
+_OSC_RE = re.compile(r"\x1B\][^\x07\x1B]*(?:\x07|\x1B\\)")
+_DCS_PM_APC_RE = re.compile(r"\x1B(?:P|_|\^)[\s\S]*?(?:\x1B\\)")
+_ESC_SINGLE_RE = re.compile(r"\x1B[@-Z\\-_]")
+_CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0B\x0C\x0E-\x1A\x1C-\x1F\x7F-\x9F]")
+_FAILURE_INDICATOR_RE = re.compile(
+    r"(?i)\b("
+    r"error|errors|failed|failure|fatal|panic|exception|traceback|"
+    r"assertion(?:error)?|segmentation fault|stack trace|"
+    r"timed out|timeout|denied"
+    r")\b|^\s*E\s+|^\s*FAILED\b|^\s*FAIL\b"
+)
+_ZERO_FAILURE_RE = re.compile(r"(?i)\b(?:0\s+failed|0\s+failures|0\s+errors?)\b")
+_POSITIVE_FAILURE_COUNT_RE = re.compile(
+    r"(?i)\b[1-9]\d*\s+(?:failed|failures|errors?)\b"
+)
+logger = logging.getLogger("src.core.services.compression_strategies")
+
+_REGEX_EVAL_SUBPROCESS_SNIPPET = (
+    "import json\n"
+    "import re\n"
+    "import sys\n"
+    "payload = json.loads(sys.stdin.read())\n"
+    "try:\n"
+    "    compiled = re.compile(payload['pattern'], payload['flags'])\n"
+    "    matched = compiled.search(payload['text']) is not None\n"
+    "except Exception:\n"
+    "    matched = False\n"
+    "sys.stdout.write('1' if matched else '0')\n"
+)
+
+
+def _line_indicates_failure(line: str) -> bool:
+    if not _FAILURE_INDICATOR_RE.search(line):
+        return False
+    return not (
+        _ZERO_FAILURE_RE.search(line) and not _POSITIVE_FAILURE_COUNT_RE.search(line)
+    )
+
+
+def _preserve_trailing_newline(*, original: str, transformed: str) -> str:
+    """Keep only trailing-newline presence aligned to the source string."""
+    if original.endswith("\n"):
+        return transformed if transformed.endswith("\n") else f"{transformed}\n"
+    return transformed.rstrip("\n")
+
+
+_GIT_CMD_ERROR_RE = re.compile(r"^(fatal|error):\s", re.MULTILINE | re.IGNORECASE)
+_GIT_CONFLICT_RE = re.compile(r"\bCONFLICT\b", re.IGNORECASE)
+_COMMIT_HASH_IN_BRACKETS_RE = re.compile(
+    r"\[([^\]\s]+)\s+([0-9a-f]{7,40})\]", re.IGNORECASE
+)
+_FILES_CHANGED_RE = re.compile(r"(\d+)\s+files?\s+changed", re.IGNORECASE)
+_INSERTIONS_RE = re.compile(r"(\d+)\s+insertions?", re.IGNORECASE)
+_DELETIONS_RE = re.compile(r"(\d+)\s+deletions?", re.IGNORECASE)
+_REF_ARROW_RE = re.compile(
+    r"^\s*([^\s]+\.{2,3}[^\s]+|[0-9a-f]{7,40}\.{2,3}[0-9a-f]{7,40})\s+(\S+)\s+->\s+(\S+)",
+    re.MULTILINE,
+)
+_NPM_ERR_RE = re.compile(r"\bERR!\b", re.IGNORECASE)
+_PIP_INSTALL_OK_RE = re.compile(
+    r"Successfully installed\s+(.+)$", re.IGNORECASE | re.MULTILINE
+)
+
+
+def _mutating_ack_failure_heuristic(text: str) -> bool:
+    if _GIT_CMD_ERROR_RE.search(text) or _GIT_CONFLICT_RE.search(text):
+        return True
+    if _NPM_ERR_RE.search(text):
+        return True
+    return any(_line_indicates_failure(line) for line in text.splitlines())