320 lines
11 KiB
Python
320 lines
11 KiB
Python
|
"""
|
|||
|
|
State store for doc-hygiene.
|
|||
|
|
|
|||
|
|
Provides:
|
|||
|
|
- resolve_project_root(start_dir, fs) — pure function, no side effects
|
|||
|
|
- StateStore — confines all writes to <project_root>/.dochygiene/
|
|||
|
|
|
|||
|
|
Design invariants honoured:
|
|||
|
|
#3 State lives in-project; no global index; never edit .gitignore.
|
|||
|
|
#4 Report rollover keeps exactly one .md + .json pair.
|
|||
|
|
#6 Deterministic-first; no model invoked here.
|
|||
|
|
#9 (scanner's concern, not ours)
|
|||
|
|
|
|||
|
|
Atomic-write mechanism: write to a temp file in the same directory,
|
|||
|
|
fsync the file descriptor, then os.replace (POSIX-atomic) onto the target.
|
|||
|
|
A concurrent reader therefore observes either the prior complete file or the
|
|||
|
|
new complete file, never a partial write.
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import tempfile
|
|||
|
|
from datetime import datetime, timezone
|
|||
|
|
from pathlib import Path
|
|||
|
|
from typing import Optional, Protocol
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# Lightweight filesystem abstraction used only by resolve_project_root
|
|||
|
|
# so the pure function can be tested without touching real disk.
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
class _RootFS(Protocol):
|
|||
|
|
"""Minimal filesystem surface for root resolution."""
|
|||
|
|
def is_dir(self, path: Path) -> bool: ...
|
|||
|
|
def parent(self, path: Path) -> Path: ...
|
|||
|
|
|
|||
|
|
|
|||
|
|
class RealRootFS:
|
|||
|
|
"""Production implementation — delegates to pathlib/os."""
|
|||
|
|
|
|||
|
|
def is_dir(self, path: Path) -> bool:
|
|||
|
|
return path.is_dir()
|
|||
|
|
|
|||
|
|
def parent(self, path: Path) -> Path:
|
|||
|
|
return path.parent
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# Pure root-resolution function (task 3.1)
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
def resolve_project_root(start_dir: Path, fs: Optional[_RootFS] = None) -> Path:
|
|||
|
|
"""
|
|||
|
|
Walk upward from *start_dir* (inclusive) looking for a .git directory.
|
|||
|
|
|
|||
|
|
Returns the first ancestor (or start_dir itself) that contains a .git
|
|||
|
|
directory. If no git root is found, returns start_dir unchanged.
|
|||
|
|
|
|||
|
|
This function is PURE: it never calls os.getcwd() or any other
|
|||
|
|
stateful function; all filesystem access goes through *fs*. Production
|
|||
|
|
callers pass os.getcwd() as start_dir; tests pass a fake fs and a
|
|||
|
|
synthetic path.
|
|||
|
|
"""
|
|||
|
|
if fs is None:
|
|||
|
|
fs = RealRootFS()
|
|||
|
|
|
|||
|
|
current = Path(start_dir)
|
|||
|
|
while True:
|
|||
|
|
if fs.is_dir(current / ".git"):
|
|||
|
|
return current
|
|||
|
|
parent = fs.parent(current)
|
|||
|
|
if parent == current:
|
|||
|
|
# Reached the filesystem root without finding .git.
|
|||
|
|
break
|
|||
|
|
current = parent
|
|||
|
|
|
|||
|
|
return Path(start_dir)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# Clock abstraction (task 3.3 / design D3)
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
class _Clock(Protocol):
|
|||
|
|
"""Returns the current UTC datetime."""
|
|||
|
|
def now(self) -> datetime: ...
|
|||
|
|
|
|||
|
|
|
|||
|
|
class RealClock:
|
|||
|
|
"""Production clock — returns the real UTC time."""
|
|||
|
|
|
|||
|
|
def now(self) -> datetime:
|
|||
|
|
return datetime.now(timezone.utc)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# StateStore (tasks 3.2–3.5)
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
_STATE_FILE = "state.json"
|
|||
|
|
_REPORT_JSON = "report.json"
|
|||
|
|
_REPORT_MD = "report.md"
|
|||
|
|
|
|||
|
|
# The set of filenames that are *not* report files and must never be deleted
|
|||
|
|
# during rollover. Explicit allowlist is safer than trying to infer.
|
|||
|
|
_NON_REPORT_FILES = {_STATE_FILE}
|
|||
|
|
|
|||
|
|
|
|||
|
|
class StateStore:
|
|||
|
|
"""
|
|||
|
|
Manages all persistent state for doc-hygiene within a single project.
|
|||
|
|
|
|||
|
|
All writes are confined to <project_root>/.dochygiene/ (invariant #3).
|
|||
|
|
No global index is maintained; each project has its own independent store.
|
|||
|
|
The store never opens or edits .gitignore (invariant #3).
|
|||
|
|
|
|||
|
|
Parameters
|
|||
|
|
----------
|
|||
|
|
project_root:
|
|||
|
|
The resolved project root directory (output of resolve_project_root).
|
|||
|
|
Injected so the store is testable with a tmp_path.
|
|||
|
|
clock:
|
|||
|
|
Provides "now". Injected for testability (design D3).
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
TIMESTAMPS = ("last_check", "last_clean", "last_reminded")
|
|||
|
|
|
|||
|
|
def __init__(self, project_root: Path, clock: Optional[_Clock] = None) -> None:
|
|||
|
|
self._root = Path(project_root)
|
|||
|
|
self._clock = clock or RealClock()
|
|||
|
|
self._state_dir = self._root / ".dochygiene"
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Directory bootstrap
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
def _ensure_state_dir(self) -> Path:
|
|||
|
|
"""Create .dochygiene/ if it does not exist. Never touches .gitignore."""
|
|||
|
|
self._state_dir.mkdir(parents=True, exist_ok=True)
|
|||
|
|
return self._state_dir
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Atomic write (task 3.4 / design D9)
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
def _atomic_write(self, target: Path, data: bytes) -> None:
|
|||
|
|
"""
|
|||
|
|
Write *data* to *target* atomically.
|
|||
|
|
|
|||
|
|
Strategy: write to a NamedTemporaryFile in the same directory,
|
|||
|
|
fsync, then os.replace onto the target. os.replace is POSIX-atomic
|
|||
|
|
within one filesystem, so a concurrent reader sees either the prior
|
|||
|
|
complete file or the new complete file, never a partial write.
|
|||
|
|
"""
|
|||
|
|
# Confirm the target is under our managed directory (confinement check).
|
|||
|
|
self._assert_confined(target)
|
|||
|
|
|
|||
|
|
target_dir = target.parent
|
|||
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|||
|
|
|
|||
|
|
fd, tmp_path_str = tempfile.mkstemp(
|
|||
|
|
dir=str(target_dir),
|
|||
|
|
prefix="." + target.name + ".tmp_",
|
|||
|
|
)
|
|||
|
|
tmp_path = Path(tmp_path_str)
|
|||
|
|
try:
|
|||
|
|
with os.fdopen(fd, "wb") as fh:
|
|||
|
|
fh.write(data)
|
|||
|
|
fh.flush()
|
|||
|
|
os.fsync(fh.fileno())
|
|||
|
|
os.replace(tmp_path_str, str(target))
|
|||
|
|
except Exception:
|
|||
|
|
# Best-effort cleanup on failure; do not mask the original error.
|
|||
|
|
try:
|
|||
|
|
tmp_path.unlink(missing_ok=True)
|
|||
|
|
except OSError:
|
|||
|
|
pass
|
|||
|
|
raise
|
|||
|
|
|
|||
|
|
def _assert_confined(self, path: Path) -> None:
|
|||
|
|
"""Raise ValueError if *path* is not under self._state_dir."""
|
|||
|
|
try:
|
|||
|
|
path.resolve().relative_to(self._state_dir.resolve())
|
|||
|
|
except ValueError:
|
|||
|
|
raise ValueError(
|
|||
|
|
f"StateStore attempted to write outside .dochygiene/: {path}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# State JSON helpers
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
def _state_path(self) -> Path:
|
|||
|
|
return self._state_dir / _STATE_FILE
|
|||
|
|
|
|||
|
|
def _read_state(self) -> dict:
|
|||
|
|
"""Read state.json; return {} if missing or unreadable (never raises)."""
|
|||
|
|
path = self._state_path()
|
|||
|
|
try:
|
|||
|
|
return json.loads(path.read_bytes())
|
|||
|
|
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
def _write_state(self, state: dict) -> None:
|
|||
|
|
self._ensure_state_dir()
|
|||
|
|
data = json.dumps(state, indent=2, sort_keys=True).encode()
|
|||
|
|
self._atomic_write(self._state_path(), data)
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Lifecycle timestamps (task 3.3)
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
def get_timestamp(self, key: str) -> Optional[datetime]:
|
|||
|
|
"""
|
|||
|
|
Return the stored datetime for *key*, or None if absent/unset.
|
|||
|
|
|
|||
|
|
The value is stored as an ISO-8601 string in state.json.
|
|||
|
|
"""
|
|||
|
|
if key not in self.TIMESTAMPS:
|
|||
|
|
raise ValueError(f"Unknown timestamp key: {key!r}")
|
|||
|
|
raw = self._read_state().get(key)
|
|||
|
|
if raw is None:
|
|||
|
|
return None
|
|||
|
|
return datetime.fromisoformat(raw)
|
|||
|
|
|
|||
|
|
def set_timestamp(self, key: str, value: Optional[datetime] = None) -> None:
|
|||
|
|
"""
|
|||
|
|
Write *key* to state.json.
|
|||
|
|
|
|||
|
|
If *value* is None, uses the injected clock's now().
|
|||
|
|
The value is serialised as an ISO-8601 string (UTC).
|
|||
|
|
"""
|
|||
|
|
if key not in self.TIMESTAMPS:
|
|||
|
|
raise ValueError(f"Unknown timestamp key: {key!r}")
|
|||
|
|
ts = value if value is not None else self._clock.now()
|
|||
|
|
# Normalise to UTC ISO-8601 string.
|
|||
|
|
if ts.tzinfo is None:
|
|||
|
|
ts = ts.replace(tzinfo=timezone.utc)
|
|||
|
|
state = self._read_state()
|
|||
|
|
state[key] = ts.isoformat()
|
|||
|
|
self._write_state(state)
|
|||
|
|
|
|||
|
|
# Convenience shorthands
|
|||
|
|
def set_last_check(self, value: Optional[datetime] = None) -> None:
|
|||
|
|
self.set_timestamp("last_check", value)
|
|||
|
|
|
|||
|
|
def get_last_check(self) -> Optional[datetime]:
|
|||
|
|
return self.get_timestamp("last_check")
|
|||
|
|
|
|||
|
|
def set_last_clean(self, value: Optional[datetime] = None) -> None:
|
|||
|
|
self.set_timestamp("last_clean", value)
|
|||
|
|
|
|||
|
|
def get_last_clean(self) -> Optional[datetime]:
|
|||
|
|
return self.get_timestamp("last_clean")
|
|||
|
|
|
|||
|
|
def set_last_reminded(self, value: Optional[datetime] = None) -> None:
|
|||
|
|
self.set_timestamp("last_reminded", value)
|
|||
|
|
|
|||
|
|
def get_last_reminded(self) -> Optional[datetime]:
|
|||
|
|
return self.get_timestamp("last_reminded")
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Report rollover (task 3.5 / design D10)
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
def _delete_existing_reports(self) -> None:
|
|||
|
|
"""
|
|||
|
|
Delete any existing report.json and report.md in .dochygiene/.
|
|||
|
|
|
|||
|
|
Only report files (report.json, report.md) are deleted.
|
|||
|
|
state.json and any other files are never touched.
|
|||
|
|
"""
|
|||
|
|
for name in (_REPORT_JSON, _REPORT_MD):
|
|||
|
|
p = self._state_dir / name
|
|||
|
|
try:
|
|||
|
|
p.unlink()
|
|||
|
|
except FileNotFoundError:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
def write_report(self, json_blob: str, md_blob: str) -> None:
|
|||
|
|
"""
|
|||
|
|
Write a new report pair, atomically, after deleting any prior pair.
|
|||
|
|
|
|||
|
|
After this call exactly one .json and one .md report file exist in
|
|||
|
|
.dochygiene/ (invariant #4).
|
|||
|
|
|
|||
|
|
Parameters
|
|||
|
|
----------
|
|||
|
|
json_blob:
|
|||
|
|
The machine-readable report JSON (as a string).
|
|||
|
|
md_blob:
|
|||
|
|
The human-readable report Markdown.
|
|||
|
|
"""
|
|||
|
|
self._ensure_state_dir()
|
|||
|
|
# Delete prior pair first (rollover).
|
|||
|
|
self._delete_existing_reports()
|
|||
|
|
# Atomically write the new pair.
|
|||
|
|
self._atomic_write(
|
|||
|
|
self._state_dir / _REPORT_JSON,
|
|||
|
|
json_blob.encode(),
|
|||
|
|
)
|
|||
|
|
self._atomic_write(
|
|||
|
|
self._state_dir / _REPORT_MD,
|
|||
|
|
md_blob.encode(),
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def read_report(self) -> Optional[tuple[str, str]]:
|
|||
|
|
"""
|
|||
|
|
Return (json_blob, md_blob) if a report exists, else None.
|
|||
|
|
"""
|
|||
|
|
json_path = self._state_dir / _REPORT_JSON
|
|||
|
|
md_path = self._state_dir / _REPORT_MD
|
|||
|
|
try:
|
|||
|
|
return json_path.read_text(), md_path.read_text()
|
|||
|
|
except FileNotFoundError:
|
|||
|
|
return None
|