cc-os/plugins/os-status/hooks/state.py

"""Per-project state for os-status (deep module).

Owns the gitignored .cc-os/ directory contract: the shared key=value config
at .cc-os/config (cross-plugin contract, ADR-026), and os-status's own state
— per-check once-per-day snooze stamps (snooze-<check>) and permanent
suppress markers (suppress-<check>) — under .cc-os/status/ (ADR-027). State
only, never code; nothing here writes outside .cc-os/. Fail-open throughout:
unreadable state reads as unset.

Pre-ADR-027 projects kept snooze-*/suppress-*/state.json at the .cc-os/ top
level, alongside config. Reads fall back to that legacy path; the first
write migrates any legacy files into .cc-os/status/ and removes them from
the old location (StateDir._migrate).
"""
from __future__ import annotations

from datetime import date
from pathlib import Path
from typing import Optional


def find_project_root(cwd: Path) -> Optional[Path]:
    """Nearest ancestor (including cwd) containing .git, else None."""
    for candidate in [cwd, *cwd.parents]:
        if (candidate / ".git").exists():
            return candidate
    return None


def read_config(root: Path) -> dict:
    """Parse .cc-os/config as '#'-commented key = value lines."""
    config = {}
    try:
        text = (root / ".cc-os" / "config").read_text()
    except Exception:
        return config
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        config[key.strip()] = value.strip()
    return config


def write_config_value(root: Path, key: str, value: str) -> None:
    """Set key=value in .cc-os/config, preserving every other line
    (comments included) and appending the key if absent. Used by
    /os-status:fix to stamp the config 'version' key.

    Normalizes to the single cross-plugin format 'key=value' — no spaces
    around '=' (issue #23) — matching os-backlog's Ruby config-write. Any
    existing key=value lines are re-emitted in the same normalized spacing
    while every other line (comments, blanks) is left untouched.
    """
    config_path = root / ".cc-os" / "config"
    try:
        lines = config_path.read_text().splitlines()
    except Exception:
        lines = []
    found = False
    out = []
    for line in lines:
        stripped = line.strip()
        if stripped and not stripped.startswith("#") and "=" in stripped:
            existing_key, existing_value = stripped.split("=", 1)
            existing_key = existing_key.strip()
            if existing_key == key:
                out.append(f"{key}={value}")
                found = True
            else:
                out.append(f"{existing_key}={existing_value.strip()}")
            continue
        out.append(line)
    if not found:
        out.append(f"{key}={value}")
    config_path.parent.mkdir(exist_ok=True)
    config_path.write_text("\n".join(out) + "\n")


LEGACY_STATE_GLOBS = ("snooze-*", "suppress-*", "state.json")


class StateDir:
    """The gitignored per-project .cc-os/status/ state directory (ADR-027).
    Reads fall back to the pre-ADR-027 .cc-os/ top level; the first write
    migrates legacy files up and removes them from the old location."""

    def __init__(self, root: Path) -> None:
        self._root = root
        self._dir = root / ".cc-os" / "status"
        self._legacy_dir = root / ".cc-os"

    def _read_path(self, name: str) -> Path:
        """The current path for `name` if it exists there, else the legacy
        top-level path (which may or may not exist — callers handle that)."""
        current = self._dir / name
        if current.exists():
            return current
        return self._legacy_dir / name

    def _migrate(self) -> None:
        """Move any legacy snooze-*/suppress-*/state.json files into
        .cc-os/status/ and remove them from .cc-os/. Idempotent; called
        before every write so migration happens transparently."""
        self._dir.mkdir(parents=True, exist_ok=True)
        for pattern in LEGACY_STATE_GLOBS:
            for legacy_file in self._legacy_dir.glob(pattern):
                if not legacy_file.is_file():
                    continue
                target = self._dir / legacy_file.name
                if not target.exists():
                    try:
                        target.write_text(legacy_file.read_text())
                    except Exception:
                        continue
                try:
                    legacy_file.unlink()
                except Exception:
                    pass

    def suppressed(self, check: str) -> bool:
        return self._read_path(f"suppress-{check}").exists()

    def snoozed(self, check: str, today: date) -> bool:
        try:
            stamped = date.fromisoformat(
                self._read_path(f"snooze-{check}").read_text().strip()
            )
        except Exception:
            return False
        return stamped == today

    def stamp(self, check: str, today: date) -> None:
        self._migrate()
        (self._dir / f"snooze-{check}").write_text(today.isoformat())

    def clear_snooze(self, check: str) -> None:
        self._migrate()
        clear_snooze(self._root, check)


def clear_snooze(root: Path, check_name: str) -> None:
    """Remove any existing snooze stamp for check_name, at both the current
    (.cc-os/status/) and legacy (.cc-os/) location. Self-healing: called by
    the SessionStart runner and /os-status:fix's checks.py --json runner
    whenever a check evaluates to ok/note, so stale snoozes from a prior
    warn don't outlive the condition they gated."""
    for base in (root / ".cc-os" / "status", root / ".cc-os"):
        try:
            (base / f"snooze-{check_name}").unlink()
        except Exception:
            pass