diff --git a/.claude/hooks/.canonical-sha256 b/.claude/hooks/.canonical-sha256 index e3514a3..5d2ad76 100644 --- a/.claude/hooks/.canonical-sha256 +++ b/.claude/hooks/.canonical-sha256 @@ -1,9 +1,9 @@ cfd43f72b3f64bde6cb779703eb13ea6dd2c55ea5ae3dace654bfa95e17345c9 security_guard.py 37ee358245e8be80b00517c32d586449cb669d6d6e02526cc37c0e6728c452d5 format_on_save.py f06a2180e64db35f96bdb896fbbfa9bf0ebc5090744817f5b87a7f0fbbb7ec61 compact_warning.py -931369475cda4f6a291ffec0fddcd99ec8eb84bef9bf2efaa46694d21ef6c740 spec_orient.py -81d06cf240c219937f349b84bba18843b83b50625b9f649b3dbdd6f87b7c8fae _state.py +c4437034774443b904e6c7cb529028ea521cf00e7a3f4859a2f964aadf08ffbc spec_orient.py +306d4d68e8e28d09f2ce102c8df6e234c2e4175b1b0c7ccf3dee6f26444584fe _state.py 63293f305ff32aab46d1da8b9d28c71ce39b658d2a8572c64024614abdf7dffe _resume_prompt.py baa145fb6fac25ae7d03a5b655b04aba25bfb77793dcdcaf44acc151394f030b _transcript_size.py 48674de791f509c539417b29214d9c87a33b7934b985597af79711ddd90ea17a _sdk_gate.py -8818f37208b6879940f6b4e43dd368f63ff3322728cdeb06bc68c073f649b345 spec_audit.py +7145a707f6e14473f71e738e3c50df059bf1ad02b3b3b9fa13e5e8b4bc247e72 spec_audit.py diff --git a/.claude/hooks/_state.py b/.claude/hooks/_state.py index ab87dbd..6e6840b 100644 --- a/.claude/hooks/_state.py +++ b/.claude/hooks/_state.py @@ -21,6 +21,7 @@ from __future__ import annotations +import json import os import re import subprocess @@ -28,6 +29,49 @@ from dataclasses import dataclass from pathlib import Path +# ── Drift-cache read (spec-status-integrity, design §3) ─────── +# +# ``spec_audit.py --pr-links`` writes ``.attune/spec-drift.json`` per +# workspace root; ``spec_orient`` reads it back so the SessionStart +# annotation stays offline (the hook's no-network invariant). Entries +# older than one weekly-CI period + slack are ignored. +_DRIFT_CACHE_MAX_AGE_SECONDS = 8 * 24 * 60 * 60 + + +def read_drift_cache(roots: list[Path], now: float | None = None) -> dict[str, dict]: + """Merge fresh drift-cache entries across workspace roots. + + Returns ``{"/": {verdict, prs, signal}}`` from every + root whose ``.attune/spec-drift.json`` is present, well-formed, and + fresher than :data:`_DRIFT_CACHE_MAX_AGE_SECONDS`. Anything else — + absent file, malformed JSON, wrong shapes, stale ``generated_at`` — + contributes nothing and never raises (the cache is advisory; the + annotation simply falls back to current behavior). + """ + current = time.time() if now is None else now + merged: dict[str, dict] = {} + for root in roots: + cache_path = Path(root) / ".attune" / "spec-drift.json" + try: + data = json.loads(cache_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError, ValueError, UnicodeDecodeError): + continue + if not isinstance(data, dict): + continue + generated = data.get("generated_at") + if not isinstance(generated, int | float): + continue + if current - float(generated) > _DRIFT_CACHE_MAX_AGE_SECONDS: + continue + specs = data.get("specs") + if not isinstance(specs, dict): + continue + for key, entry in specs.items(): + if isinstance(key, str) and isinstance(entry, dict): + merged.setdefault(key, entry) + return merged + + # Phases checked, highest-priority first. The first phase file # present in a spec directory determines the displayed phase # and status. @@ -59,7 +103,7 @@ # done (see decisions.md DECIDE-2). _TERMINAL_LINE = re.compile( r"^\s*\**\s*(?:Spec\s+)?Status\**\s*:\s*\**\s*" - r"(closed|complete|completed|retired|superseded|shipped|done)\b", + r"(closed|complete|completed|retired|superseded|shipped|done|implemented)\b", re.IGNORECASE | re.MULTILINE, ) _CHECKLIST_HEADING = re.compile( @@ -76,12 +120,30 @@ ) _NEXT_H2 = re.compile(r"^##\s+", re.MULTILINE) _TERMINAL_VERDICTS = frozenset( - {"closed", "complete", "completed", "retired", "superseded", "shipped", "done"} + { + "closed", + "complete", + "completed", + "retired", + "superseded", + "shipped", + "done", + # spec-status-integrity additions (design §1): ``implemented`` + # and the bare check glyphs are terminal — the wild-scan found + # both in real headers. + "implemented", + "✓", + "✅", + } ) # Ongoing-by-design statuses: a living roadmap / continuous program is not # pending work, so it should NOT show as in-flight — but it is also not # "done". Excluded from the in-flight list, kept distinct from terminal. _ONGOING_VERDICTS = frozenset({"living", "ongoing"}) +# Parked-semantics statuses (spec-status-integrity, design §1): work +# deliberately on hold. Skipped by drift checks and excluded from the +# in-flight orientation list; the audit still lists them (as parked). +_PARKED_VERDICTS = frozenset({"parked", "paused", "blocked", "deferred"}) # Leading alphabetic word of a status value, for first-word tokenization: # ``complete (2026-06-09) — shipped #694`` -> ``complete``. @@ -96,13 +158,164 @@ def _leading_verdict(status: str) -> str: not exact-string membership. This is the fix for the class of bug where a correctly-marked-``complete`` spec stayed in-flight forever because ``"complete (date) — ..."`` is not in ``_TERMINAL_VERDICTS``. + + A bare check glyph (``✓`` / ``✅``) leading the value is returned + as-is — the glyphs are members of ``_TERMINAL_VERDICTS`` + (spec-status-integrity, design §1) but are not alphabetic, so the + word regex would otherwise skip past them to whatever word follows. """ + stripped = status.strip().lstrip("*_`").strip() + for glyph in ("✅", "✓"): + if stripped.startswith(glyph): + return glyph # search (not match) so a stray leading ``**``/punctuation doesn't # swallow the verdict — the first alphabetic run is the word we want. m = _LEADING_WORD.search(status) return m.group(0).lower() if m else "" +# ── Status-vocabulary lint (spec-status-integrity, design §1) ─ +# +# The canonical vocabulary for NEW specs. Historical aliases +# (``_TERMINAL_VERDICTS`` / ``_ONGOING_VERDICTS`` members and the +# in-flight forms below) stay accepted — no mass rewrite of ~127 +# existing status lines — but the lint steers authors to these 8. +_CANONICAL_STATUS_TOKENS = ( + "draft", + "in-review", + "approved", + "in-progress", + "implemented", + "complete", + "superseded", + "parked", +) +# In-flight forms honored by the checker (design §2): the canonical +# in-flight 4 plus the historical ``not started`` / ``open`` / ``pending``. +# ``not`` is the leading token of ``not started``. +_IN_FLIGHT_TOKENS = frozenset( + {"draft", "in-review", "approved", "in-progress", "not", "open", "pending"} +) +# Lint tokenization keeps hyphens (``in-review``), unlike +# ``_LEADING_WORD`` which stops at the first non-alpha char. +_LINT_TOKEN = re.compile(r"[A-Za-z][A-Za-z-]*") + + +def lint_status_token(status: str) -> str | None: + """Lint a status value's leading token against the known vocabulary. + + Returns ``None`` when the leading token is recognized — one of the + canonical 8, a historical terminal/ongoing alias, a parked-family + token, an accepted in-flight form, or a bare check glyph. Otherwise + returns a one-line ``unparseable`` message naming the canonical 8; + the token is never guessed at. + """ + stripped = status.strip().lstrip("*_`").strip() + for glyph in ("✅", "✓"): + if stripped.startswith(glyph): + return None + match = _LINT_TOKEN.search(status) + token = match.group(0).lower() if match else "" + recognized = ( + _TERMINAL_VERDICTS | _ONGOING_VERDICTS | _PARKED_VERDICTS | _IN_FLIGHT_TOKENS + ) | set(_CANONICAL_STATUS_TOKENS) + if token in recognized: + return None + shown = token or status.strip() or "(empty)" + return f'unparseable status "{shown}" — use one of: ' + ", ".join(_CANONICAL_STATUS_TOKENS) + + +# ── PR-reference extraction (spec-status-integrity, design §1) ── +# +# ``extract_pr_refs()`` parses the four PR-citation styles found in the +# wild (workspace spec design.md §1): explicit ``PR #212`` / +# ``PRs #303, #304`` lists, bare ``#1191`` (ambiguous — may be an +# issue; resolved at check time via the pulls API, merged-only), and +# the markdown pull-URL — the required style for cross-repo refs. +# Code fences and inline code spans are blanked first so quoted +# examples (`` `#NNN` `` in docs) never count as citations. +_CODE_FENCE = re.compile(r"```.*?```", re.DOTALL) +_INLINE_CODE = re.compile(r"`[^`\n]+`") +_PULL_URL = re.compile(r"https://github\.com/(?P[\w.-]+/[\w.-]+)/pull/(?P\d+)\b") +# Whole markdown links are blanked AFTER pull-URLs are extracted, so a +# link text like ``[#95](…/pull/95)`` doesn't ALSO scan as a bare +# current-repo ref — and issue-links yield nothing at all. +_MD_LINK = re.compile(r"\[[^\]\n]*\]\([^)\n]*\)") +_PR_LIST = re.compile(r"\bPRs?\s*#\d+(?:\s*,\s*#\d+)*", re.IGNORECASE) +_REF_NUMBER = re.compile(r"#(\d+)") +# Bare ``#NNN``: not preceded by a word char / ``#`` / ``/`` (rejects +# ``foo#12``, anchors, URL paths) and not followed by a word char +# (rejects ``#12abc``). +_BARE_REF = re.compile(r"(? str: + """Length-preserving mask so match positions stay comparable.""" + return " " * (match.end() - match.start()) + + +def extract_pr_refs(text: str) -> list[PrRef]: + """Extract PR citations from spec text — deduped, document order. + + Styles recognized (workspace design §1): ``PR #212``, + ``PRs #303, #304``, bare ``#1191``, and + ``https://github.com///pull/`` (markdown-wrapped or + bare). Duplicate ``(repo, number)`` pairs collapse to the earliest + occurrence, upgraded to ``explicit`` if any occurrence was. + """ + scrubbed = _CODE_FENCE.sub(_blank, text) + scrubbed = _INLINE_CODE.sub(_blank, scrubbed) + + hits: list[tuple[int, str | None, int, bool]] = [] # (pos, repo, number, explicit) + + for match in _PULL_URL.finditer(scrubbed): + hits.append((match.start(), match.group("repo"), int(match.group("number")), True)) + # Blank markdown links wholesale (pull-URLs already harvested), then + # any bare pull-URLs outside links, before the plain-text scans. + scrubbed = _MD_LINK.sub(_blank, scrubbed) + scrubbed = _PULL_URL.sub(_blank, scrubbed) + + for match in _PR_LIST.finditer(scrubbed): + for num in _REF_NUMBER.finditer(match.group(0)): + hits.append((match.start() + num.start(), None, int(num.group(1)), True)) + scrubbed = _PR_LIST.sub(_blank, scrubbed) + + for match in _BARE_REF.finditer(scrubbed): + hits.append((match.start(), None, int(match.group(1)), False)) + + # Dedupe on (repo, number): earliest position wins the slot; the + # explicit flag is OR-merged across occurrences. + best: dict[tuple[str | None, int], tuple[int, bool]] = {} + for pos, repo, number, explicit in hits: + key = (repo, number) + if key in best: + prev_pos, prev_explicit = best[key] + best[key] = (prev_pos, prev_explicit or explicit) + else: + best[key] = (pos, explicit) + ordered = sorted(best.items(), key=lambda item: item[1][0]) + return [ + PrRef(number=number, repo=repo, explicit=explicit) + for (repo, number), (_pos, explicit) in ordered + ] + + # ── Deliverables block (spec-status-integrity, DECIDE-3) ────── # # A machine-readable "## Deliverables" section names the paths/globs @@ -516,7 +729,9 @@ def classify_staleness(spec_text: str, header_status: str, roots: list[Path]) -> # done deeper in the file is not falsely flagged. effective, _source, _conflict = _reconcile_status(header_status, spec_text) lead = _leading_verdict(effective) - if lead in _TERMINAL_VERDICTS or lead in _ONGOING_VERDICTS: + # Parked-family statuses are deliberately on hold — skipped by drift + # checks (design §1), so they classify ``ok`` alongside terminal. + if lead in _TERMINAL_VERDICTS or lead in _ONGOING_VERDICTS or lead in _PARKED_VERDICTS: return "ok" return "suspected-stale" @@ -577,6 +792,9 @@ def _is_in_flight(phase: str, effective_status: str) -> bool: phase. - First word is ongoing-by-design (living / ongoing) → a continuous program / living roadmap is not pending work, exclude. + - First word is parked-family (parked / paused / blocked / + deferred) → deliberately on hold, not pending work, exclude + (spec-status-integrity, design §1). - Empty status (malformed) → still in-flight (don't drop a working spec because the heading was malformed). - Anything else (draft / approved / in-progress / …) → in-flight. @@ -586,7 +804,7 @@ def _is_in_flight(phase: str, effective_status: str) -> bool: (the self-truthing improvement). """ lead = _leading_verdict(effective_status) - if lead in _TERMINAL_VERDICTS or lead in _ONGOING_VERDICTS: + if lead in _TERMINAL_VERDICTS or lead in _ONGOING_VERDICTS or lead in _PARKED_VERDICTS: return False return True diff --git a/.claude/hooks/spec_audit.py b/.claude/hooks/spec_audit.py index 93b2bbc..d428ef7 100644 --- a/.claude/hooks/spec_audit.py +++ b/.claude/hooks/spec_audit.py @@ -8,9 +8,24 @@ Spec | Layer | Status | Staleness | Unresolved D-7 — **warn by default, gate opt-in.** Exits ``0`` even when stale -specs exist; ``--strict`` exits ``1`` on any ``suspected-stale`` so a -repo can wire a hard CI gate. Crash-proof: any unexpected error prints -what we have and still exits ``0`` (warn-by-default never hard-fails). +specs exist; ``--strict`` exits ``1`` on any ``suspected-stale`` (or, +with ``--pr-links``, any ``drifted``) so a repo can wire a hard CI +gate. Crash-proof: any unexpected error prints what we have and still +exits ``0`` (warn-by-default never hard-fails). + +spec-status-integrity additions (workspace design §2): + +- ``--pr-links`` — the PRIMARY drift signal. For each in-flight spec, + extract PR citations from its phase files and resolve them via + ``gh`` (merged PRs only; bounded + cached per run). ≥1 merged + implementing PR + in-flight status ⇒ **drifted**. Results are + written to ``/.attune/spec-drift.json`` for the offline + ``spec_orient`` annotation. +- ``--offline`` — skip all ``gh`` calls (the deliverable-existence + signal still runs). A missing/failing ``gh`` degrades the same way — + the audit never blocks on network state. +- ``--json`` — machine-readable output for the weekly CI's + tracking-issue upsert. Run via ``make spec-audit`` or ``python plugin/hooks/spec_audit.py``. @@ -20,7 +35,10 @@ from __future__ import annotations +import json +import subprocess import sys +import time import traceback from dataclasses import dataclass from pathlib import Path @@ -37,25 +55,153 @@ sys.path.insert(0, _HOOKS_DIR) from _state import ( # noqa: E402 — sys.path bootstrap above + PrRef, + _is_in_flight, _resolve_entry, discover_specs, + extract_pr_refs, workspace_roots, ) -# Display order — suspected-stale rows surface first; ``ok`` sinks last. +# Display order — drifted (merged-PR evidence) outranks even +# suspected-stale; ``ok`` sinks last. _STALENESS_ORDER = { - "suspected-stale": 0, - "partial": 1, - "unknown": 2, - "docs-only": 3, - "opted-out": 4, - "ok": 5, + "drifted": 0, + "suspected-stale": 1, + "partial": 2, + "unknown": 3, + "docs-only": 4, + "opted-out": 5, + "ok": 6, } -# Only suspected-stale gets a glyph; the rest render verbatim. -_STALENESS_LABEL = {"suspected-stale": "⚠ suspected-stale"} +# Only the actionable verdicts get a glyph; the rest render verbatim. +_STALENESS_LABEL = {"suspected-stale": "⚠ suspected-stale", "drifted": "⚠ drifted"} _HEADERS = ("Spec", "Layer", "Status", "Staleness", "Unresolved") +# ── PR-link resolution (workspace design §2) ────────────────── + +# Bound gh calls per run — same discipline as session_recall.py's +# _MAX_PR_CHECKS, sized for an audit sweep rather than a session start. +_MAX_GH_CALLS = 30 +_GH_TIMEOUT_SECONDS = 6.0 +# All three phase files are scanned for citations, not just the +# highest-priority one — a requirements.md often carries the approval +# trail ("shipped in #N") even when tasks.md exists. +_PHASE_FILENAMES = ("tasks.md", "design.md", "requirements.md") + + +def _run_gh(args: list[str], cwd: Path | None) -> subprocess.CompletedProcess[str] | None: + """Invoke ``gh`` — the subprocess boundary tests patch. + + Returns ``None`` when the binary is missing or the call times out; + callers treat that as "unknown", never as evidence. + """ + try: + return subprocess.run( # noqa: S603, S607 — fixed argv, no shell + ["gh", *args], + capture_output=True, + text=True, + timeout=_GH_TIMEOUT_SECONDS, + cwd=str(cwd) if cwd else None, + ) + except FileNotFoundError: + raise + except (OSError, subprocess.TimeoutExpired): + return None + + +class _PrResolver: + """Resolves PR refs to merged/not-merged via ``gh``, bounded + cached. + + - Results are cached per ``(repo-or-layer, number)`` for the run, so + a PR cited from three phase files costs one call. + - At most ``max_calls`` gh invocations per run; past the cap every + further ref is "unknown" (not merged) — conservative, no false + drift claims. + - A missing ``gh`` binary marks the resolver dead: all subsequent + lookups short-circuit to "unknown" (offline degradation, design + §2 — never blocks). + """ + + def __init__(self, max_calls: int = _MAX_GH_CALLS) -> None: + self.calls = 0 + self.dead = False + self._cache: dict[tuple[str, int], bool] = {} + + def merged(self, ref: PrRef, spec_dir: Path, layer: str) -> bool: + """True iff ``ref`` resolves to a MERGED pull request.""" + key = (ref.repo or f"local:{layer}", ref.number) + if key in self._cache: + return self._cache[key] + if self.dead or self.calls >= _MAX_GH_CALLS: + return False + self.calls += 1 + try: + verdict = self._check(ref, spec_dir) + except FileNotFoundError: + # gh binary absent — go dead for the rest of the run + # (offline degradation; the deliverable signal still stands). + self.dead = True + return False + self._cache[key] = verdict + return verdict + + def _check(self, ref: PrRef, spec_dir: Path) -> bool: + # Explicit cross-repo slug → REST pulls endpoint ("merged" is a + # single-PR-GET field; an issue number 404s here, which is + # exactly the merged-only filter the design wants). Local refs + # resolve through the spec dir's own git context, so the right + # repo is picked per layer without any hardcoded slug map. + if ref.repo: + proc = _run_gh(["api", f"repos/{ref.repo}/pulls/{ref.number}"], cwd=None) + else: + proc = _run_gh( + ["pr", "view", str(ref.number), "--json", "state"], + cwd=spec_dir, + ) + if proc is None or proc.returncode != 0: + return False + try: + data = json.loads(proc.stdout) + except (json.JSONDecodeError, ValueError): + return False + if not isinstance(data, dict): + return False + if ref.repo: + return bool(data.get("merged")) + return data.get("state") == "MERGED" + + +def _collect_refs(spec_dir: Path) -> list[PrRef]: + """Union of PR citations across a spec's phase files, deduped.""" + seen: set[tuple[str | None, int]] = set() + refs: list[PrRef] = [] + for fname in _PHASE_FILENAMES: + fpath = spec_dir / fname + if not fpath.is_file(): + continue + try: + text = fpath.read_text(encoding="utf-8", errors="replace") + except OSError: + continue + for ref in extract_pr_refs(text): + key = (ref.repo, ref.number) + if key in seen: + continue + seen.add(key) + refs.append(ref) + return refs + + +def _pr_label(ref: PrRef, layer: str) -> str: + """Human label for a merged PR — ``attune-author #95`` / ``#67``.""" + if ref.repo: + return f"{ref.repo.rsplit('/', 1)[-1]} #{ref.number}" + if layer and layer != "workspace": + return f"{layer} #{ref.number}" + return f"#{ref.number}" + @dataclass(frozen=True) class AuditResult: @@ -67,14 +213,45 @@ class AuditResult: staleness: str resolved: int # entries that resolve on disk total: int # entries declared + # spec-status-integrity additions — defaults keep older callers + # (and the pre-existing tests) constructing rows unchanged. + in_flight: bool = False + drifted: bool = False + merged_prs: tuple[str, ...] = () + signal: str = "deliverables" + cache_key: str = "" # "/" — the drift-cache key + root: str = "" # workspace root the spec lives under + +def _root_for(spec_path: Path, roots: list[Path]) -> str: + """The workspace root ``spec_path`` lives under (no symlink resolve — + spec paths are built by prefixing the root, so prefix-match holds).""" + for root in roots: + try: + if spec_path.is_relative_to(root): + return str(root) + except (TypeError, ValueError): + continue + return "" -def audit_specs(roots: list[Path] | None = None) -> list[AuditResult]: + +def audit_specs( + roots: list[Path] | None = None, + *, + pr_links: bool = False, + resolver: _PrResolver | None = None, +) -> list[AuditResult]: """Classify every discovered spec (terminal included) into a row. Resolution counts are recomputed per spec so the report can show how many declared deliverables are present. Per-spec failures degrade to a zero-count row rather than aborting the whole audit. + + With ``pr_links`` (and a resolver), each IN-FLIGHT spec's phase + files are additionally scanned for PR citations; ≥1 merged + implementing PR marks the row ``drifted`` (workspace design §2). + Terminal/parked specs are never queried — no gh spend on settled + specs. """ if roots is None: roots = workspace_roots() @@ -89,6 +266,20 @@ def audit_specs(roots: list[Path] | None = None) -> list[AuditResult]: resolved = sum(1 for e in spec.deliverables if _resolve_entry(e, roots)) except Exception: # noqa: BLE001 — one bad spec must not abort the audit resolved = 0 + in_flight = _is_in_flight(spec.phase, spec.effective_status) + drifted = False + merged_prs: tuple[str, ...] = () + checked = pr_links and resolver is not None and in_flight + if checked: + try: + merged_prs = tuple( + _pr_label(ref, spec.layer) + for ref in _collect_refs(spec.path) + if resolver.merged(ref, spec.path, spec.layer) + ) + except Exception: # noqa: BLE001 — one bad spec must not abort the audit + merged_prs = () + drifted = bool(merged_prs) results.append( AuditResult( slug=spec.slug, @@ -97,11 +288,81 @@ def audit_specs(roots: list[Path] | None = None) -> list[AuditResult]: staleness=spec.staleness, resolved=resolved, total=total, + in_flight=in_flight, + drifted=drifted, + merged_prs=merged_prs, + signal="pr-links" if checked else "deliverables", + cache_key=f"{spec.layer}/{spec.slug}", + root=_root_for(spec.path, roots), ) ) return results +def write_drift_cache(results: list[AuditResult], roots: list[Path]) -> list[Path]: + """Write ``.attune/spec-drift.json`` under each workspace root. + + Schema (workspace design §2): + ``{generated_at, specs: {"/": {verdict, prs, signal}}}``. + Only in-flight specs are recorded — terminal/parked rows carry no + drift signal. A clean (empty) specs map is still written so + ``spec_orient`` can tell "checked and clean" from "never checked". + Per-root write failures are swallowed: the cache is an optimization, + never a blocker. + """ + now = time.time() + written: list[Path] = [] + for root in roots: + entries = { + r.cache_key: { + "verdict": "drifted" if r.drifted else "ok", + "prs": list(r.merged_prs), + "signal": r.signal, + } + for r in results + if r.in_flight and r.root == str(root) + } + cache_path = Path(root) / ".attune" / "spec-drift.json" + try: + cache_path.parent.mkdir(parents=True, exist_ok=True) + cache_path.write_text( + json.dumps({"generated_at": now, "specs": entries}, indent=2) + "\n", + encoding="utf-8", + ) + except OSError: + continue + written.append(cache_path) + return written + + +def format_json(results: list[AuditResult]) -> str: + """Machine-readable audit payload for the tracking-issue upsert.""" + drifted = sorted(r.cache_key for r in results if r.drifted) + payload = { + "generated_at": time.time(), + "counts": { + "specs": len(results), + "drifted": len(drifted), + "suspected_stale": sum(1 for r in results if r.staleness == "suspected-stale"), + }, + "drifted": drifted, + "specs": { + r.cache_key: { + "layer": r.layer, + "slug": r.slug, + "status": r.status, + "staleness": r.staleness, + "in_flight": r.in_flight, + "drifted": r.drifted, + "prs": list(r.merged_prs), + "signal": r.signal, + } + for r in results + }, + } + return json.dumps(payload, indent=2) + + def _truncate(text: str, limit: int) -> str: """Clip ``text`` to ``limit`` chars, ellipsizing the overflow.""" return text if len(text) <= limit else text[: limit - 1] + "…" @@ -109,6 +370,8 @@ def _truncate(text: str, limit: int) -> str: def _detail(result: AuditResult) -> str: """Render the ``Unresolved`` column for one row.""" + if result.drifted: + return ", ".join(result.merged_prs) + " merged" if result.staleness == "opted-out": return "(opt-out)" if result.staleness == "docs-only": @@ -121,15 +384,23 @@ def _detail(result: AuditResult) -> str: return "—" +def _verdict(result: AuditResult) -> str: + """Displayed verdict — merged-PR evidence outranks the staleness + heuristic for the same row.""" + return "drifted" if result.drifted else result.staleness + + def format_report(results: list[AuditResult]) -> str: """Render the full audit matrix as a string.""" - stale = [r for r in results if r.staleness == "suspected-stale"] + stale = [r for r in results if r.staleness == "suspected-stale" and not r.drifted] + drifted = [r for r in results if r.drifted] noun = "spec" if len(results) == 1 else "specs" - title = f"SPEC STATUS AUDIT — {len(results)} {noun} ({len(stale)} suspected-stale)" + counts = f"{len(drifted)} drifted, {len(stale)} suspected-stale" + title = f"SPEC STATUS AUDIT — {len(results)} {noun} ({counts})" if not results: return f"{title}\n\n(no specs found)" - ordered = sorted(results, key=lambda r: (_STALENESS_ORDER.get(r.staleness, 9), r.slug)) + ordered = sorted(results, key=lambda r: (_STALENESS_ORDER.get(_verdict(r), 9), r.slug)) rows = [ ( _truncate(r.slug, 44), @@ -138,7 +409,7 @@ def format_report(results: list[AuditResult]) -> str: # whole paragraph — truncate so one long status doesn't blow # the column out (e.g. integration-coverage's 1k-char line). _truncate(r.status, 32), - _STALENESS_LABEL.get(r.staleness, r.staleness), + _STALENESS_LABEL.get(_verdict(r), _verdict(r)), _detail(r), ) for r in ordered @@ -151,24 +422,38 @@ def _line(cells: tuple[str, ...]) -> str: out = [title, "", _line(_HEADERS), "─" * len(_line(_HEADERS))] out.extend(_line(row) for row in rows) out.append("") + if drifted: + out.append( + f"⚠ {len(drifted)} spec(s) have merged implementing PRs but an " + "in-flight status — flip the status or mark them parked." + ) if stale: out.append( f"⚠ {len(stale)} spec(s) have shipped deliverables but a " "non-terminal status — verify & update." ) - else: - out.append("✓ No suspected-stale specs.") + if not drifted and not stale: + out.append("✓ No drifted or suspected-stale specs.") return "\n".join(out) def main(argv: list[str] | None = None) -> int: - """Print the audit matrix; exit per ``--strict``. Never hard-crashes.""" + """Print the audit report; exit per ``--strict``. Never hard-crashes.""" try: args = sys.argv[1:] if argv is None else argv strict = "--strict" in args - results = audit_specs() - print(format_report(results)) - if strict and any(r.staleness == "suspected-stale" for r in results): + offline = "--offline" in args + pr_links = "--pr-links" in args and not offline + as_json = "--json" in args + roots = workspace_roots() + resolver = _PrResolver() if pr_links else None + results = audit_specs(roots, pr_links=pr_links, resolver=resolver) + if pr_links: + # The cache is what lets the offline SessionStart hook + # annotate drift without network calls (design §3). + write_drift_cache(results, roots) + print(format_json(results) if as_json else format_report(results)) + if strict and any(r.drifted or r.staleness == "suspected-stale" for r in results): return 1 return 0 except Exception: # noqa: BLE001 — warn-by-default: report and exit 0 diff --git a/.claude/hooks/spec_orient.py b/.claude/hooks/spec_orient.py index 7b25305..1f8c84a 100644 --- a/.claude/hooks/spec_orient.py +++ b/.claude/hooks/spec_orient.py @@ -24,6 +24,7 @@ from __future__ import annotations import json +import os import sys import traceback from pathlib import Path @@ -46,9 +47,21 @@ SpecInfo, discover_specs, prune_stale_sentinels, + read_drift_cache, workspace_roots, ) + +def _audit_annotations_enabled() -> bool: + """Kill switch (spec-status-integrity, design §3). + + ``ATTUNE_SPEC_AUDIT=off`` suppresses both the ``⚠ drifted`` + drift-cache annotation and the existing suspected-stale hint. + Anything else (including unset) leaves them on. + """ + return os.environ.get("ATTUNE_SPEC_AUDIT", "").strip().lower() != "off" + + # Char budget for the post-compact spec body. Generous so the # spec survives compaction with full content; the model sees this # as fresh context immediately after the compact summary. @@ -58,9 +71,13 @@ _ORIENTATION_MAX_SPECS = 3 -def _format_phase(spec: SpecInfo) -> str: +def _format_phase(spec: SpecInfo, annotate: bool = True) -> str: """Short ``(phase status)`` blurb for the orientation list. + ``annotate=False`` (the ``ATTUNE_SPEC_AUDIT=off`` kill switch) + suppresses the suspected-stale hint; the status_conflict hint is + self-truthing, not an audit annotation, and always renders. + Renders the reconciled ``effective_status`` (not the raw header ``status``) so a stale "draft" header above a closed checklist doesn't show up as in-flight draft. @@ -91,20 +108,40 @@ def _format_phase(spec: SpecInfo) -> str: }.get(spec.status_source, spec.status_source) raw = spec.status or "no header" return f'{base} — {source_label}; header says "{raw}", worth fixing' - if spec.staleness == "suspected-stale": + if annotate and spec.staleness == "suspected-stale": raw = spec.status or "no status" - return f'{base} — ⚠ deliverables present, status still "{raw}"; ' "verify before building" + return f'{base} — ⚠ deliverables present, status still "{raw}"; verify before building' return base -def format_orientation(specs: list[SpecInfo]) -> str: +def _drift_suffix(spec: SpecInfo, drift_cache: dict[str, dict]) -> str: + """`` ⚠ drifted: attune-author #95 merged`` when the drift cache + marks this spec drifted; empty otherwise (design §3). The cache is + read from disk only — the hook makes no network calls.""" + entry = drift_cache.get(f"{spec.layer}/{spec.slug}") + if not isinstance(entry, dict) or entry.get("verdict") != "drifted": + return "" + prs = [p for p in entry.get("prs") or [] if isinstance(p, str)] + detail = f": {', '.join(prs)} merged" if prs else "" + return f" ⚠ drifted{detail}" + + +def format_orientation( + specs: list[SpecInfo], + drift_cache: dict[str, dict] | None = None, + annotate: bool = True, +) -> str: """Short markdown list of in-flight specs for non-compact starts.""" if not specs: return "" + drift = drift_cache or {} lines = ["attune workspace — in-flight specs:"] for spec in specs[:_ORIENTATION_MAX_SPECS]: layer_prefix = "" if spec.layer == "workspace" else f"{spec.layer}/" - lines.append(f"- {layer_prefix}specs/{spec.slug}/ ({_format_phase(spec)})") + line = f"- {layer_prefix}specs/{spec.slug}/ ({_format_phase(spec, annotate=annotate)})" + if annotate: + line += _drift_suffix(spec, drift) + lines.append(line) leftover = len(specs) - _ORIENTATION_MAX_SPECS if leftover > 0: lines.append(f"- (+{leftover} more)") @@ -167,7 +204,11 @@ def main() -> int: if body: print(body) else: - orient = format_orientation(specs) + annotate = _audit_annotations_enabled() + # Offline by design: the drift signal comes from the cache + # spec_audit --pr-links wrote, never from a network call. + drift_cache = read_drift_cache(roots) if annotate else {} + orient = format_orientation(specs, drift_cache=drift_cache, annotate=annotate) if orient: print(orient) return 0 diff --git a/.github/workflows/spec-status-reminder.yml b/.github/workflows/spec-status-reminder.yml new file mode 100644 index 0000000..da7c1e6 --- /dev/null +++ b/.github/workflows/spec-status-reminder.yml @@ -0,0 +1,16 @@ +# Thin caller — canonical logic lives in attune-ai (reusable workflow). +# spec-status-integrity task 7; see attune workspace +# specs/spec-status-integrity/. +name: spec-status-reminder + +on: + pull_request: + types: [closed] + +permissions: + contents: read + pull-requests: write + +jobs: + remind: + uses: Smart-AI-Memory/attune-ai/.github/workflows/spec-status-reminder.yml@main