Class PhaseTracker

ci/meson/phase_tracker.py:18–172

Thread-safe phase state tracker for build and test execution. Tracks the current phase (CONFIGURE/COMPILE/LINK/EXECUTE) to help disambiguate build and test failures. Uses atomic file writes and stale detection to handle process crashes gracefully. Example: tracker = Ph

class PhaseTracker:
    """
    Thread-safe phase state tracker for build and test execution.

    Tracks the current phase (CONFIGURE/COMPILE/LINK/EXECUTE) to help
    disambiguate build and test failures. Uses atomic file writes and
    stale detection to handle process crashes gracefully.

    Example:
        tracker = PhaseTracker(build_dir, "quick")
        tracker.set_phase("CONFIGURE")
        # ... meson setup ...
        tracker.set_phase("COMPILE", target="test_async")
        # ... compilation ...
        tracker.set_phase("EXECUTE", test_name="test_async")
        # ... test execution ...
        tracker.clear()  # Success
    """

    def __init__(self, build_dir: Path, build_mode: str):
        """
        Initialize PhaseTracker.

        Args:
            build_dir: Meson build directory (e.g., .build/meson-quick)
            build_mode: Build mode name (e.g., "quick", "debug", "release")
        """
        self.build_dir = Path(build_dir)
        self.build_mode = build_mode
        self.state_dir = self.build_dir.parent / "meson-state"
        self.state_file = self.state_dir / f"current-phase-{build_mode}.json"
        self._lock = threading.Lock()

        # Ensure state directory exists
        self.state_dir.mkdir(parents=True, exist_ok=True)

        # Register cleanup handler for process exit
        self._register_cleanup()

    def set_phase(
        self,
        phase: Literal["CONFIGURE", "COMPILE", "LINK", "EXECUTE"],
        test_name: Optional[str] = None,
        target: Optional[str] = None,
        path: Literal["sequential", "streaming"] = "sequential",
    ) -> None:
        """
        Update phase state (thread-safe, atomic write).

        Args:
            phase: Current build/test phase
            test_name: Name of test being executed (optional)
            target: Compilation target (optional)
            path: Execution path ("sequential" or "streaming")
        """
        with self._lock:
            state = {
                "phase": phase,