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

"""Status checks for the os-status plugin (deep module).

Each check is a pure function of Ctx returning a CheckResult; REGISTRY is the
single list the SessionStart runner iterates. Adding a check is one function
plus one Check entry here — no discovery, no subprocess protocol (design D1).

PRESENT_NOTE and ABSENT_NOTE are copied byte-identical from os-adr's
hooks/session_start.py, the wording source of record (Eval-B-tuned; see
invariants.md). Do not paraphrase them.
"""
from __future__ import annotations

import json
import os
import re
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import Callable, Optional

PRESENT_NOTE = (
    "[os-adr] This project records architecture decisions in docs/adr/ "
    "(index: docs/adr/README.md). WHEN your task involves making an "
    "architecture-level choice, or changing/replacing an approach that "
    "already exists in this codebase (persistence, protocols, libraries, "
    "conventions, error handling) -> FIRST run /os-adr:find to check whether "
    "a decision already covers it. Mechanical rule: BEFORE your first edit "
    "to any existing source or config file -> run /os-adr:find on the paths "
    "you are about to touch (one cheap deterministic CLI call; additions "
    "count — a new method can bypass a decided constraint). WHEN "
    "you make such a choice -> record it "
    "with /os-adr:create. If find shows an Accepted ADR covering the approach "
    "you are changing, you are reversing a recorded decision: the task is "
    "NOT COMPLETE until the superseding ADR is created with /os-adr:create "
    "(supersedes: NNNN) — code changed but ADR unrecorded means the decision "
    "history now asserts the opposite of what the code does."
)

ABSENT_NOTE = (
    "[os-adr] No ADR system in this project. /os-adr:init sets one up; "
    "/os-adr:migrate converts existing decision docs. "
    "(touch .os-adr/suppress to silence this permanently)"
)

ENV_VAR = "CLAUDE_CODE_SUBAGENT_MODEL"

DEFAULT_VAULT_PATH = "~/Documents/SecondBrain"
VAULT_SKIP_DIRS = {"graphify-out", "_templates", ".graphify"}


@dataclass(frozen=True)
class Ctx:
    """Everything a check may read: project root (None outside a git
    project), the user's settings.json path, the vault path, per-project
    .cc-os/config, and the process environment."""

    project_root: Optional[Path]
    settings_path: Path
    vault_path: Path
    config: dict
    environ: dict


@dataclass(frozen=True)
class CheckResult:
    status: str  # "ok" | "note" | "warn"
    message: str = ""


OK = CheckResult("ok")


def note(message: str) -> CheckResult:
    return CheckResult("note", message)


def warn(message: str) -> CheckResult:
    return CheckResult("warn", message)


def _settings_env(path: Path) -> dict:
    try:
        env = json.loads(path.read_text()).get("env") or {}
        return env if isinstance(env, dict) else {}
    except Exception:
        return {}


def subagent_model_env_override(ctx: Ctx) -> CheckResult:
    """Warn when CLAUDE_CODE_SUBAGENT_MODEL silently forces every subagent
    to one model (the WS1 Cluster 1 incident). Runs outside git projects too."""
    found = []
    if ENV_VAR in ctx.environ:
        found.append(f"the process environment (={ctx.environ[ENV_VAR]})")
    settings_env = _settings_env(ctx.settings_path)
    if ENV_VAR in settings_env:
        found.append(f"the env block of {ctx.settings_path} (={settings_env[ENV_VAR]})")
    if not found:
        return OK
    return warn(
        f"{ENV_VAR} is set in " + " and in ".join(found)
        + " — every subagent spawn is silently forced to that model, overriding"
        " explicit Agent model params. Remove it unless intended."
    )


def adr_system_present(ctx: Ctx) -> CheckResult:
    """Port of os-adr's SessionStart behavior: present -> the verbatim usage
    note; absent -> init/migrate suggestion (runner snoozes it once per day),
    honoring the legacy .os-adr/suppress marker."""
    root = ctx.project_root
    adr_dir = root / "docs" / "adr"
    if adr_dir.is_dir() and (adr_dir / "README.md").is_file():
        return note(PRESENT_NOTE)
    if (root / ".os-adr" / "suppress").exists():
        return OK
    return warn(ABSENT_NOTE)


def _vault_notes(vault: Path):
    for path in vault.rglob("*.md"):
        parts = path.relative_to(vault).parts
        if any(p.startswith(".") or p in VAULT_SKIP_DIRS for p in parts[:-1]):
            continue
        yield path


def _is_hub_for(text: str, project: str) -> bool:
    return bool(
        re.search(r"(?m)^\s*-\s*type/hub\s*$", text)
        and re.search(rf"(?m)^\s*-\s*project/{re.escape(project)}\s*$", text)
    )


def vault_hub_note_present(ctx: Ctx) -> CheckResult:
    """An explicit hub slug in .cc-os/config wins; otherwise infer by scanning
    vault frontmatter for type/hub + project/<dirname> facet tags."""
    vault = ctx.vault_path
    if not vault.is_dir():
        return OK  # no vault on this machine — nothing to check
    slug = str(ctx.config.get("hub", "")).strip()
    if slug:
        if (vault / f"{slug}.md").is_file():
            return OK
        return warn(
            f".cc-os/config names hub note '{slug}' but {slug}.md does not"
            f" exist in the vault ({vault}) — fix the slug or create the note"
            " with /os-vault:write."
        )
    project = ctx.project_root.name
    for path in _vault_notes(vault):
        try:
            head = path.read_text(errors="replace")[:2048]
        except Exception:
            continue
        if _is_hub_for(head, project):
            return OK
    return warn(
        f"No vault hub note found for project '{project}' — run /os-vault:write"
        f" to create one (tags: type/hub + project/{project}), or set"
        " 'hub = <slug>' in .cc-os/config if one exists under another name."
    )


AUDIT_LEDGER_ENV = "OS_ORCH_AUDIT_LEDGER"
DEFAULT_AUDIT_LEDGER = "~/.local/state/os-orchestration/audit/metrics.tsv"
AUDIT_INTERVAL_DAYS = 14


def _last_audit_date(ledger: Path):
    try:
        rows = [l for l in ledger.read_text().splitlines() if l.strip()][1:]
        return date.fromisoformat(rows[-1].split("\t")[0])
    except Exception:
        return None


def orchestration_audit_due(ctx: Ctx) -> CheckResult:
    """Nudge when the biweekly orchestration IRL audit is overdue. The ledger
    (metrics.tsv) is appended deterministically by os-orchestration's
    audit/bin/audit-stats; this check only reads its last row's date."""
    ledger = Path(
        os.path.expanduser(ctx.environ.get(AUDIT_LEDGER_ENV, DEFAULT_AUDIT_LEDGER))
    )
    last = _last_audit_date(ledger)
    if last is None:
        return OK  # no ledger — this machine hasn't opted into audits yet
    days = (date.today() - last).days
    if days < AUDIT_INTERVAL_DAYS:
        return OK
    return warn(
        f"orchestration IRL audit due (last run {last.isoformat()}, {days} days"
        " ago) — run /os-orchestration:audit-sessions (works from any project;"
        " deterministic stats precompute, then auditor fan-out)."
    )


TRACKER_ABSENT_NOTE = (
    "No 'tracker' key set in .cc-os/config — this project's issue tracker is"
    " unregistered. Run /os-backlog:route to pick a tracker and"
    " record it (planka:<board> | forgejo:<owner>/<repo> |"
    " github:<owner>/<repo> | repo:<path>)."
)

# --- config version (ADR-026) ---------------------------------------------

CURRENT_CONFIG_VERSION = 1


def config_version_current(ctx: Ctx) -> CheckResult:
    """Warn when .cc-os/config has no 'version' key or an older one than
    CURRENT_CONFIG_VERSION — the project was set up under an older cc-os
    approach. /os-status:fix stamps the current version."""
    raw = str(ctx.config.get("version", "")).strip()
    try:
        version = int(raw)
    except ValueError:
        version = None
    if version is not None and version >= CURRENT_CONFIG_VERSION:
        return OK
    return warn(
        f".cc-os/config has no current 'version' key (found: {raw or 'unset'},"
        f" current: {CURRENT_CONFIG_VERSION}) — run /os-status:fix to bring"
        " this project up to date."
    )


# --- project graph (ADR-026) -----------------------------------------------


def project_graph_present(ctx: Ctx) -> CheckResult:
    """Warn when the project has no Graphify codebase graph at its root."""
    if (ctx.project_root / "graphify-out" / "graph.json").is_file():
        return OK
    return warn(
        "No project Graphify graph found (graphify-out/graph.json) — run"
        " /os-vault:onboard-project to build one."
    )

_TRACKER_PATTERNS = {
    "planka": re.compile(r"^[^/\s]+$"),
    "forgejo": re.compile(r"^[^/\s]+/[^/\s]+$"),
    "github": re.compile(r"^[^/\s]+/[^/\s]+$"),
    "repo": re.compile(r"^\S+$"),
}


def tracker_configured(ctx: Ctx) -> CheckResult:
    """Read the 'tracker' key from .cc-os/config. Present + valid -> silent
    (configured projects cost nothing at session start); present + malformed
    -> one-line warn, never a crash; absent -> daily-snoozed nudge toward the
    os-backlog routing skill."""
    raw = str(ctx.config.get("tracker", "")).strip()
    if not raw:
        return warn(TRACKER_ABSENT_NOTE)
    scheme, sep, rest = raw.partition(":")
    pattern = _TRACKER_PATTERNS.get(scheme)
    if not sep or not pattern or not pattern.match(rest):
        return warn(
            f".cc-os/config sets tracker = '{raw}', which does not match the"
            " expected grammar (planka:<board> | forgejo:<owner>/<repo> |"
            " github:<owner>/<repo> | repo:<path>) — fix the value."
        )
    return OK


@dataclass(frozen=True)
class Check:
    name: str
    fn: Callable
    project_scoped: bool
    remediation: str = ""


REGISTRY = [
    Check(
        "subagent-model-env-override",
        subagent_model_env_override,
        project_scoped=False,
        remediation="remove CLAUDE_CODE_SUBAGENT_MODEL from the environment / settings.json",
    ),
    Check(
        "adr-system-present",
        adr_system_present,
        project_scoped=True,
        remediation="/os-adr:init",
    ),
    Check(
        "vault-hub-note-present",
        vault_hub_note_present,
        project_scoped=True,
        remediation="/os-vault:write (create a hub note) or set hub= in .cc-os/config",
    ),
    Check(
        "orchestration-audit-due",
        orchestration_audit_due,
        project_scoped=False,
        remediation="/os-orchestration:audit-sessions",
    ),
    Check(
        "tracker-configured",
        tracker_configured,
        project_scoped=True,
        remediation="/os-backlog:route",
    ),
    Check(
        "config-version-current",
        config_version_current,
        project_scoped=True,
        remediation="/os-status:fix",
    ),
    Check(
        "project-graph-present",
        project_graph_present,
        project_scoped=True,
        remediation="/os-vault:onboard-project",
    ),
]


# --- JSON runner (ADR-026) --------------------------------------------------
#
# `python3 hooks/checks.py --json` runs REGISTRY against the cwd project and
# prints a machine-readable JSON array for the /os-status:fix skill. This is
# additive: it does not change how session_start.py imports from this module.


def build_ctx() -> "Ctx":
    from state import find_project_root, read_config  # local import, no cycle

    root = find_project_root(Path.cwd())
    config = read_config(root) if root else {}
    return Ctx(
        project_root=root,
        settings_path=Path.home() / ".claude" / "settings.json",
        vault_path=Path(os.path.expanduser(config.get("vault_path", DEFAULT_VAULT_PATH))),
        config=config,
        environ=dict(os.environ),
    )


def run_all_json(ctx: "Ctx") -> list:
    """Run every applicable check and return a list of plain dicts:
    {name, status, message, remediation}. project_scoped checks are skipped
    when ctx.project_root is None, same rule as the SessionStart runner."""
    results = []
    for check in REGISTRY:
        if check.project_scoped and ctx.project_root is None:
            continue
        try:
            result = check.fn(ctx)
            if not isinstance(result, CheckResult):
                raise TypeError("check returned a non-CheckResult")
        except Exception:
            result = CheckResult("warn", f"status check '{check.name}' failed to run")
        results.append(
            {
                "name": check.name,
                "status": result.status,
                "message": result.message,
                "remediation": check.remediation,
            }
        )
    return results


def _main_json() -> int:
    print(json.dumps(run_all_json(build_ctx())))
    return 0


if __name__ == "__main__":
    import sys

    if "--json" in sys.argv[1:]:
        raise SystemExit(_main_json())