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)."
    )


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


REGISTRY = [
    Check("subagent-model-env-override", subagent_model_env_override, project_scoped=False),
    Check("adr-system-present", adr_system_present, project_scoped=True),
    Check("vault-hub-note-present", vault_hub_note_present, project_scoped=True),
    Check("orchestration-audit-due", orchestration_audit_due, project_scoped=False),
]