""" State store for doc-hygiene. Provides: - resolve_project_root(start_dir, fs) — pure function, no side effects - StateStore — confines all writes to /.cc-os/dochygiene/ Per ADR-027, the canonical state directory is /.cc-os/dochygiene/. The legacy path /.dochygiene/ is still supported for reads: if the canonical directory has no state, reads fall back to the legacy directory. On the first write, any existing legacy files are migrated into the canonical directory and the legacy directory is removed. 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 shutil 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 /.cc-os/dochygiene/ (invariant #3, ADR-027). No global index is maintained; each project has its own independent store. The store never opens or edits .gitignore (invariant #3). Backward compatibility (ADR-027): the legacy location /.dochygiene/ is read as a fallback when the canonical directory has no state. The first write migrates any legacy files into the canonical directory and removes the legacy directory. 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 / ".cc-os" / "dochygiene" self._legacy_dir = self._root / ".dochygiene" # ------------------------------------------------------------------ # Directory bootstrap # ------------------------------------------------------------------ def _ensure_state_dir(self) -> Path: """Create .cc-os/dochygiene/ if it does not exist, migrating any legacy .dochygiene/ contents in first. Never touches .gitignore.""" self._migrate_legacy() self._state_dir.mkdir(parents=True, exist_ok=True) return self._state_dir def _migrate_legacy(self) -> None: """ One-time migration of /.dochygiene/ into /.cc-os/dochygiene/. No-op if the legacy directory does not exist. Copies every file found in the legacy directory into the canonical directory (canonical files, if any already exist, take precedence and are not overwritten), then removes the legacy directory entirely. """ if not self._legacy_dir.is_dir(): return self._state_dir.mkdir(parents=True, exist_ok=True) for entry in self._legacy_dir.iterdir(): if not entry.is_file(): continue target = self._state_dir / entry.name if not target.exists(): target.write_bytes(entry.read_bytes()) shutil.rmtree(self._legacy_dir, ignore_errors=True) # ------------------------------------------------------------------ # 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 .cc-os/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). Falls back to the legacy .dochygiene/state.json (ADR-027) if the canonical file is absent. The legacy file is only *read* here — it is migrated into the canonical location on the next write. """ try: return json.loads(self._state_path().read_bytes()) except (FileNotFoundError, json.JSONDecodeError, OSError): pass try: return json.loads((self._legacy_dir / _STATE_FILE).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. Falls back to the legacy .dochygiene/ directory (ADR-027) if the canonical directory has no report pair. """ try: return ( (self._state_dir / _REPORT_JSON).read_text(), (self._state_dir / _REPORT_MD).read_text(), ) except FileNotFoundError: pass try: return ( (self._legacy_dir / _REPORT_JSON).read_text(), (self._legacy_dir / _REPORT_MD).read_text(), ) except FileNotFoundError: return None