395 lines
13 KiB
Python
395 lines
13 KiB
Python
"""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-context/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-context'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-context: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 (forgejo:<owner>/<repo> |"
|
|
" github:<owner>/<repo> | repo:<path>)."
|
|
)
|
|
|
|
TRACKER_PLANKA_RETIRED_NOTE = (
|
|
"planka: trackers are retired (ADR-0042 — git issues are the single"
|
|
" tracker for state and specs). Run /os-backlog:route to re-route to"
|
|
" forgejo:/github:/repo:."
|
|
)
|
|
|
|
# --- 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 = {
|
|
"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; a retired `planka:` value -> one-line
|
|
warn naming ADR-0042; absent -> daily-snoozed nudge toward the
|
|
os-backlog routing skill. Always fail-soft: never blocks the session."""
|
|
raw = str(ctx.config.get("tracker", "")).strip()
|
|
if not raw:
|
|
return warn(TRACKER_ABSENT_NOTE)
|
|
scheme, sep, rest = raw.partition(":")
|
|
if scheme == "planka":
|
|
return warn(TRACKER_PLANKA_RETIRED_NOTE)
|
|
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 (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-context: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.
|
|
|
|
Self-healing (issue #23): whenever a check evaluates to ok/note, clear
|
|
any stale snooze file for it, same as the SessionStart runner — so a
|
|
/os-status:fix run cleans up snoozes left behind by checks it just
|
|
fixed, not just checks the next SessionStart happens to re-run.
|
|
"""
|
|
from state import clear_snooze # local import, no cycle
|
|
|
|
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")
|
|
if result.status in ("ok", "note") and ctx.project_root is not None:
|
|
clear_snooze(ctx.project_root, check.name)
|
|
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())
|