454 lines
16 KiB
Python
454 lines
16 KiB
Python
"""Dependency-free Windows/WSL environment diagnostics for repository gates."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
from collections.abc import Callable
|
|
from contextlib import suppress
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Final, Literal
|
|
from urllib.parse import quote
|
|
|
|
DIAGNOSTIC_SCHEMA: Final = "sigmax.environment-diagnostics/1"
|
|
PlatformId = Literal["windows", "linux"]
|
|
CacheIntegrity = Literal["ok", "corrupt", "locked"]
|
|
FilesystemStatus = Literal["ok", "locked", "cleanup_failed"]
|
|
OptionalLane = Literal["none", "plot", "reference"]
|
|
|
|
_OPTIONAL_MODULES: Final[dict[OptionalLane, tuple[str, ...]]] = {
|
|
"none": (),
|
|
"plot": ("matplotlib",),
|
|
"reference": ("diffusers",),
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True, kw_only=True)
|
|
class EnvironmentIssue:
|
|
"""One stable environment failure with a direct recovery action."""
|
|
|
|
code: str
|
|
summary: str
|
|
remediation: str
|
|
|
|
def projection(self) -> dict[str, str]:
|
|
return {
|
|
"code": self.code,
|
|
"remediation": self.remediation,
|
|
"summary": self.summary,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True, kw_only=True)
|
|
class EnvironmentObservation:
|
|
"""Pure observations consumed by the policy evaluator."""
|
|
|
|
platform: PlatformId
|
|
local_venv: bool
|
|
cache_contained: bool
|
|
cache_writable: bool
|
|
cache_integrity: CacheIntegrity
|
|
precommit_candidate_count: int
|
|
precommit_foreign: bool
|
|
filesystem_status: FilesystemStatus
|
|
unicode_roundtrip: bool
|
|
temp_contained: bool
|
|
temp_writable: bool
|
|
anonymous_temp_compatible: bool
|
|
temp_capture_mitigated: bool
|
|
optional_lane: OptionalLane
|
|
missing_optional_modules: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True, kw_only=True)
|
|
class EnvironmentDiagnosticReport:
|
|
"""Deterministic diagnostic result without private absolute paths."""
|
|
|
|
platform: PlatformId
|
|
optional_lane: OptionalLane
|
|
issues: tuple[EnvironmentIssue, ...]
|
|
mitigations: tuple[str, ...]
|
|
|
|
@property
|
|
def status(self) -> str:
|
|
return "FAIL" if self.issues else "PASS"
|
|
|
|
def projection(self) -> dict[str, object]:
|
|
return {
|
|
"issues": [issue.projection() for issue in self.issues],
|
|
"mitigations": list(self.mitigations),
|
|
"optional_lane": self.optional_lane,
|
|
"platform": self.platform,
|
|
"schema": DIAGNOSTIC_SCHEMA,
|
|
"status": self.status,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True, kw_only=True)
|
|
class EnvironmentProbeConfig:
|
|
"""Resolved runtime inputs for a bounded live probe."""
|
|
|
|
repository_root: Path
|
|
environment_prefix: Path
|
|
expected_environment: Path
|
|
cache_root: Path
|
|
temp_root: Path
|
|
platform: PlatformId
|
|
optional_lane: OptionalLane
|
|
path_value: str
|
|
temp_capture_mitigated: bool = False
|
|
|
|
|
|
def _venv_label(platform: PlatformId) -> str:
|
|
return ".venv" if platform == "windows" else ".venv-wsl"
|
|
|
|
|
|
def _cache_label(platform: PlatformId) -> str:
|
|
return ".tmp/pre-commit-win" if platform == "windows" else ".tmp/pre-commit-linux"
|
|
|
|
|
|
def _temp_label(platform: PlatformId) -> str:
|
|
return ".tmp/runtime-win" if platform == "windows" else ".tmp/runtime-linux"
|
|
|
|
|
|
def _venv_install_command(platform: PlatformId) -> str:
|
|
if platform == "windows":
|
|
return 'python -m venv .venv; .venv\\Scripts\\python.exe -m pip install -e ".[dev]"'
|
|
return "python3 -m venv .venv-wsl && .venv-wsl/bin/python -m pip install -e '.[dev]'"
|
|
|
|
|
|
def evaluate_environment(observation: EnvironmentObservation) -> EnvironmentDiagnosticReport:
|
|
"""Classify collected observations into stable actionable failures."""
|
|
|
|
issues: list[EnvironmentIssue] = []
|
|
cache_label = _cache_label(observation.platform)
|
|
temp_label = _temp_label(observation.platform)
|
|
|
|
if not observation.local_venv:
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="venv.not_local",
|
|
summary=f"The active interpreter is not owned by {_venv_label(observation.platform)}.",
|
|
remediation=(
|
|
"Create it and install '.[dev]' before retrying. Run: "
|
|
f"{_venv_install_command(observation.platform)}."
|
|
),
|
|
)
|
|
)
|
|
if not observation.cache_contained:
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="cache.outside_workspace",
|
|
summary="The configured pre-commit cache is outside repository .tmp.",
|
|
remediation=(
|
|
f"Set PRE_COMMIT_HOME to {cache_label} and rerun with the local venv."
|
|
),
|
|
)
|
|
)
|
|
elif not observation.cache_writable:
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="cache.not_writable",
|
|
summary="The repository-local pre-commit cache is not writable.",
|
|
remediation=(
|
|
f"Close hook processes, restore write permission on {cache_label}, and retry."
|
|
),
|
|
)
|
|
)
|
|
if observation.cache_integrity == "corrupt":
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="cache.corrupt",
|
|
summary="The repository-local pre-commit SQLite manifest is corrupt.",
|
|
remediation=(
|
|
f"Close pre-commit, remove only {cache_label}, then rerun to rebuild the cache."
|
|
),
|
|
)
|
|
)
|
|
elif observation.cache_integrity == "locked":
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="cache.locked",
|
|
summary="The repository-local pre-commit SQLite manifest is locked.",
|
|
remediation=(
|
|
"Close other pre-commit/Python processes, verify antivirus exclusions for "
|
|
f"{cache_label}, and retry serially."
|
|
),
|
|
)
|
|
)
|
|
if observation.precommit_candidate_count > 1 or observation.precommit_foreign:
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="tooling.precommit_conflict",
|
|
summary="PATH exposes multiple or non-venv pre-commit executables.",
|
|
remediation=(
|
|
"Remove conflicting PATH entries and run the selected local interpreter with "
|
|
"python -m pre_commit."
|
|
),
|
|
)
|
|
)
|
|
if observation.filesystem_status == "locked":
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="filesystem.locked",
|
|
summary="The gate cannot complete an owned file write/replace/delete cycle.",
|
|
remediation=(
|
|
"Close processes holding the file, inspect antivirus or sync-tool locks, restore "
|
|
"permissions, and rerun the preflight."
|
|
),
|
|
)
|
|
)
|
|
elif observation.filesystem_status == "cleanup_failed":
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="filesystem.cleanup_failed",
|
|
summary="The owned environment probe directory could not be removed.",
|
|
remediation=(
|
|
f"Close file handles, remove only the probe directory below {temp_label}, and retry."
|
|
),
|
|
)
|
|
)
|
|
if not observation.unicode_roundtrip:
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="unicode.roundtrip_failed",
|
|
summary="The filesystem failed a non-ASCII create/read/rename round-trip.",
|
|
remediation=(
|
|
"Use a Unicode-capable filesystem and UTF-8 locale, then recreate the local venv "
|
|
"and rerun."
|
|
),
|
|
)
|
|
)
|
|
if not observation.temp_contained:
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="temp.outside_workspace",
|
|
summary="The configured gate temp directory is outside repository .tmp.",
|
|
remediation=f"Set SIGMAX_TEMP_ROOT, TMPDIR, TMP, and TEMP to {temp_label}.",
|
|
)
|
|
)
|
|
elif not observation.temp_writable:
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="temp.not_writable",
|
|
summary="The repository-local gate temp directory is not writable.",
|
|
remediation=(
|
|
f"Create {temp_label}, restore write permission, and rerun the local wrapper."
|
|
),
|
|
)
|
|
)
|
|
if not observation.anonymous_temp_compatible and not observation.temp_capture_mitigated:
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="temp.incompatible",
|
|
summary="The configured temp path cannot sustain anonymous file operations.",
|
|
remediation=(
|
|
"On WSL mounted workspaces, run the canonical Linux wrapper so pytest uses "
|
|
"SIGMAX_PYTEST_CAPTURE_MODE=sys; otherwise choose a compatible writable temp path."
|
|
),
|
|
)
|
|
)
|
|
if observation.missing_optional_modules:
|
|
modules = ", ".join(observation.missing_optional_modules)
|
|
issues.append(
|
|
EnvironmentIssue(
|
|
code="optional.missing",
|
|
summary=f"Optional lane {observation.optional_lane} is missing: {modules}.",
|
|
remediation=(
|
|
"Install the reviewed extra with the selected local interpreter: "
|
|
f'python -m pip install -e ".[{observation.optional_lane}]".'
|
|
),
|
|
)
|
|
)
|
|
mitigations = (
|
|
("pytest.capture_sys",)
|
|
if not observation.anonymous_temp_compatible and observation.temp_capture_mitigated
|
|
else ()
|
|
)
|
|
return EnvironmentDiagnosticReport(
|
|
platform=observation.platform,
|
|
optional_lane=observation.optional_lane,
|
|
issues=tuple(issues),
|
|
mitigations=mitigations,
|
|
)
|
|
|
|
|
|
def _is_within(path: Path, parent: Path) -> bool:
|
|
try:
|
|
path.resolve().relative_to(parent.resolve())
|
|
except (OSError, ValueError):
|
|
return False
|
|
return True
|
|
|
|
|
|
def _probe_writable_directory(path: Path) -> bool:
|
|
probe: Path | None = None
|
|
try:
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
descriptor, name = tempfile.mkstemp(prefix=".sigmax-write-", dir=path)
|
|
os.close(descriptor)
|
|
probe = Path(name)
|
|
probe.write_bytes(b"sigmax")
|
|
return probe.read_bytes() == b"sigmax"
|
|
except OSError:
|
|
return False
|
|
finally:
|
|
if probe is not None:
|
|
with suppress(OSError):
|
|
probe.unlink(missing_ok=True)
|
|
|
|
|
|
def _cache_integrity(cache_root: Path) -> CacheIntegrity:
|
|
database = cache_root / "db.db"
|
|
if not database.is_file():
|
|
return "ok"
|
|
# IMPORTANT: use SQLite read-only mode; diagnostics must never repair or recreate a cache DB.
|
|
uri = f"file:{quote(str(database.resolve()))}?mode=ro"
|
|
try:
|
|
connection = sqlite3.connect(uri, uri=True, timeout=0)
|
|
try:
|
|
result = connection.execute("PRAGMA quick_check").fetchone()
|
|
finally:
|
|
connection.close()
|
|
except sqlite3.Error as error:
|
|
lowered = str(error).lower()
|
|
return "locked" if "locked" in lowered or "busy" in lowered else "corrupt"
|
|
return "ok" if result == ("ok",) else "corrupt"
|
|
|
|
|
|
def _find_precommit_candidates(path_value: str, platform: PlatformId) -> tuple[Path, ...]:
|
|
names = (
|
|
("pre-commit.exe", "pre-commit.cmd", "pre-commit.bat", "pre-commit")
|
|
if platform == "windows"
|
|
else ("pre-commit",)
|
|
)
|
|
found: dict[str, Path] = {}
|
|
for raw_directory in path_value.split(os.pathsep):
|
|
if not raw_directory:
|
|
continue
|
|
directory = Path(raw_directory)
|
|
for name in names:
|
|
candidate = directory / name
|
|
if candidate.is_file():
|
|
try:
|
|
resolved = candidate.resolve()
|
|
except OSError:
|
|
resolved = candidate.absolute()
|
|
found[os.path.normcase(str(resolved))] = resolved
|
|
return tuple(found[key] for key in sorted(found))
|
|
|
|
|
|
def _probe_file_operations(temp_root: Path) -> tuple[FilesystemStatus, bool, bool, bool]:
|
|
probe_directory: Path | None = None
|
|
source: Path | None = None
|
|
target: Path | None = None
|
|
filesystem_status: FilesystemStatus = "ok"
|
|
unicode_roundtrip = True
|
|
temp_writable = True
|
|
anonymous_temp_compatible = True
|
|
try:
|
|
temp_root.mkdir(parents=True, exist_ok=True)
|
|
probe_directory = Path(tempfile.mkdtemp(prefix=".sigmax-environment-", dir=temp_root))
|
|
source = probe_directory / "probe.txt"
|
|
target = probe_directory / "環境-探測.txt"
|
|
source.write_text("Sigmax 路徑", encoding="utf-8")
|
|
if source.read_text(encoding="utf-8") != "Sigmax 路徑":
|
|
unicode_roundtrip = False
|
|
source.replace(target)
|
|
if target.read_text(encoding="utf-8") != "Sigmax 路徑":
|
|
unicode_roundtrip = False
|
|
target.unlink()
|
|
target = None
|
|
try:
|
|
with tempfile.TemporaryFile(dir=temp_root) as anonymous:
|
|
anonymous.write(b"sigmax")
|
|
anonymous.seek(0)
|
|
if anonymous.read() != b"sigmax":
|
|
anonymous_temp_compatible = False
|
|
anonymous.truncate()
|
|
except OSError:
|
|
anonymous_temp_compatible = False
|
|
except UnicodeError:
|
|
unicode_roundtrip = False
|
|
except OSError:
|
|
filesystem_status = "locked"
|
|
temp_writable = False
|
|
finally:
|
|
cleanup_failed = False
|
|
for candidate in (target, source):
|
|
if candidate is not None:
|
|
try:
|
|
candidate.unlink(missing_ok=True)
|
|
except OSError:
|
|
cleanup_failed = True
|
|
if probe_directory is not None:
|
|
try:
|
|
probe_directory.rmdir()
|
|
except OSError:
|
|
cleanup_failed = True
|
|
if cleanup_failed:
|
|
filesystem_status = "cleanup_failed"
|
|
return filesystem_status, unicode_roundtrip, temp_writable, anonymous_temp_compatible
|
|
|
|
|
|
def collect_environment_observation(
|
|
config: EnvironmentProbeConfig,
|
|
*,
|
|
module_available: Callable[[str], bool] | None = None,
|
|
file_operation_probe: Callable[[Path], tuple[FilesystemStatus, bool, bool, bool]] | None = None,
|
|
) -> EnvironmentObservation:
|
|
"""Collect bounded observations without importing optional modules."""
|
|
|
|
owned_root = config.repository_root / ".tmp"
|
|
cache_contained = _is_within(config.cache_root, owned_root)
|
|
temp_contained = _is_within(config.temp_root, owned_root)
|
|
# CRITICAL: never touch cache/temp paths until containment below repository .tmp is proven.
|
|
cache_writable = cache_contained and _probe_writable_directory(config.cache_root)
|
|
cache_integrity: CacheIntegrity = (
|
|
_cache_integrity(config.cache_root) if cache_contained else "ok"
|
|
)
|
|
if temp_contained:
|
|
probe = file_operation_probe or _probe_file_operations
|
|
(
|
|
filesystem_status,
|
|
unicode_roundtrip,
|
|
temp_writable,
|
|
anonymous_temp_compatible,
|
|
) = probe(config.temp_root)
|
|
else:
|
|
filesystem_status = "ok"
|
|
unicode_roundtrip = True
|
|
temp_writable = False
|
|
anonymous_temp_compatible = True
|
|
|
|
candidates = _find_precommit_candidates(config.path_value, config.platform)
|
|
precommit_foreign = any(
|
|
not _is_within(candidate, config.expected_environment) for candidate in candidates
|
|
)
|
|
available = module_available or (lambda name: importlib.util.find_spec(name) is not None)
|
|
missing_optional = tuple(
|
|
name for name in _OPTIONAL_MODULES[config.optional_lane] if not available(name)
|
|
)
|
|
local_venv = _is_within(config.environment_prefix, config.expected_environment)
|
|
|
|
return EnvironmentObservation(
|
|
platform=config.platform,
|
|
local_venv=local_venv,
|
|
cache_contained=cache_contained,
|
|
cache_writable=cache_writable,
|
|
cache_integrity=cache_integrity,
|
|
precommit_candidate_count=len(candidates),
|
|
precommit_foreign=precommit_foreign,
|
|
filesystem_status=filesystem_status,
|
|
unicode_roundtrip=unicode_roundtrip,
|
|
temp_contained=temp_contained,
|
|
temp_writable=temp_writable,
|
|
anonymous_temp_compatible=anonymous_temp_compatible,
|
|
temp_capture_mitigated=config.temp_capture_mitigated,
|
|
optional_lane=config.optional_lane,
|
|
missing_optional_modules=missing_optional,
|
|
)
|