chore(hooks): re-sync — worktree-drift alarm + PR-ref precision (attune-ai #1317)

.claude/hooks/.canonical-sha256

@@ -1,10 +1,10 @@
 b62c7991281cf1cc0c34c38f5338fa85eb9c08e7163859d7771e57d0e8e2359c  security_guard.py
 0983dd23febbae7d2b429d1aa7b21df387811427a664b044a3b19d8821c05dd8  format_on_save.py
 0def63915a6bfb0b023a02e504281535ab19d7e2e1e0604901cdb68e345f3990  compact_warning.py
-7f924f99fb331b9f30a77f253a69ee4820e2a4f4f257062d58b22530d56ba41a  spec_orient.py
-d52f97f551a5c8068df2c6afb6faa6b18be795189a9c604d99bd99b815f19315  _state.py
+80a8634dcba4ee3cabd062015144257e8424c732115f0921de744037294128b8  spec_orient.py
+848d870f4f2770151aa26e2301eb5a100e226a58c7b2ec8d323b887aa1ff84fe  _state.py
 63293f305ff32aab46d1da8b9d28c71ce39b658d2a8572c64024614abdf7dffe  _resume_prompt.py
 baa145fb6fac25ae7d03a5b655b04aba25bfb77793dcdcaf44acc151394f030b  _transcript_size.py
 48674de791f509c539417b29214d9c87a33b7934b985597af79711ddd90ea17a  _sdk_gate.py
-68743464283f0d19f7aca1a491cb122daed037462dd69d7bc3bb1dfe56ae84bb  spec_audit.py
+2d1ba09af11b3eb36d370fa31d3bf438536acf025ea4586587977908e5826c25  spec_audit.py
 76bedca6b34a9126b4bd8d864afad52c6ec72229b26b51f00ff8b4ab673474eb  _bootstrap.py

.claude/hooks/_state.py

@@ -247,6 +247,23 @@ def lint_status_token(status: str) -> str | None:
 # ``foo#12``, anchors, URL paths) and not followed by a word char
 # (rejects ``#12abc``).
 _BARE_REF = re.compile(r"(?<![\w#/])#(\d+)(?![\w#])")
+# Shipping vocabulary that promotes a bare ref to citation confidence.
+# First live sweep (attune-ai#1314) showed bare ``#N`` matching ordinal
+# prose ("failure-shape #1", "open item #2") and resolving to unrelated
+# ancient PRs; a bare ref only reads as a PR citation when its own line
+# says something shipped through it ("shipped in #67", "landed via #95").
+_SHIPPING_CONTEXT = re.compile(
+    r"\b(ship(?:ped|s|ping)?|merged?|land(?:ed|s)?|implement(?:ed|s|ing|ation)?"
+    r"|deliver(?:ed|s)?|released?|fix(?:ed|es)?|via|clos(?:ed|es))\b",
+    re.IGNORECASE,
+)
+
+
+def _line_around(text: str, pos: int) -> str:
+    """The full line of ``text`` containing offset ``pos``."""
+    start = text.rfind("\n", 0, pos) + 1
+    end = text.find("\n", pos)
+    return text[start : end if end != -1 else len(text)]
 
 
 @dataclass(frozen=True)
@@ -255,9 +272,11 @@ class PrRef:
 
     ``repo`` is the explicit ``owner/name`` slug from a pull-URL
     citation; ``None`` means "current repo". ``explicit`` is True when
-    the citation is unambiguously a PR (``PR #N`` / ``PRs #N, …`` /
-    pull-URL); a bare ``#NNN`` is ``explicit=False`` — it may be an
-    issue, which the checker resolves via the pulls API at check time.
+    the citation reads as a PR with confidence: ``PR #N`` /
+    ``PRs #N, …`` / pull-URL, or a bare ``#NNN`` whose own line carries
+    shipping vocabulary ("shipped in #67"). Other bare refs stay
+    ``explicit=False`` — ordinal prose ("failure-shape #1") must not
+    drive the drift signal (attune-ai#1314).
     """
 
     number: int
@@ -297,7 +316,11 @@ def extract_pr_refs(text: str) -> list[PrRef]:
     scrubbed = _PR_LIST.sub(_blank, scrubbed)
 
     for match in _BARE_REF.finditer(scrubbed):
-        hits.append((match.start(), None, int(match.group(1)), False))
+        # attune-ai#1314: a bare ref earns citation confidence only when
+        # its line carries shipping vocabulary; ordinal prose stays
+        # explicit=False and the drift signal ignores it.
+        confident = bool(_SHIPPING_CONTEXT.search(_line_around(scrubbed, match.start())))
+        hits.append((match.start(), None, int(match.group(1)), confident))
 
     # Dedupe on (repo, number): earliest position wins the slot; the
     # explicit flag is OR-merged across occurrences.
@@ -931,6 +954,110 @@ def _run_git(cwd: Path, *args: str) -> str:
     return result.stdout
 
 
+# ── Worktree spec-drift alarm (spec-status-integrity task 10) ──
+#
+# Motivating incident: approved spec content sat uncommitted in a
+# worktree for six weeks while the SessionStart hook read those local
+# files and reported the spec "approved" — masking the drift. CI can
+# never see local worktrees, so this scan must live in the local hook.
+
+_WORKTREE_DRIFT_AGE_DAYS = 7
+_MAX_WORKTREES_SCANNED = 12
+_WORKTREE_SCAN_BUDGET_SECONDS = 2.5
+_SPEC_STATUS_PATHSPECS = ("specs", "docs/specs")
+
+
+@dataclass(frozen=True)
+class WorktreeSpecDrift:
+    """Uncommitted spec content sitting in one worktree past the age gate."""
+
+    worktree: str  # basename of the worktree directory
+    since: str  # YYYY-MM-DD of the oldest stale file's mtime
+    sample: str  # one repo-relative path, for the alarm line
+    count: int  # total stale spec files in this worktree
+
+
+def _stale_spec_files(worktree: Path, cutoff: float) -> list[tuple[float, str]]:
+    """(mtime, relpath) for uncommitted spec files older than ``cutoff``."""
+    porcelain = _run_git(
+        worktree, "status", "--porcelain", "--no-renames", "--", *_SPEC_STATUS_PATHSPECS
+    )
+    stale: list[tuple[float, str]] = []
+    for line in porcelain.splitlines():
+        rel = line[3:].strip().strip('"')
+        if not rel:
+            continue
+        path = worktree / rel
+        candidates = [path]
+        if rel.endswith("/") or path.is_dir():
+            # Untracked directories are reported as one entry; look one
+            # level of rglob deep enough to date the content (bounded).
+            candidates = [f for f in path.rglob("*") if f.is_file()][:50]
+        for f in candidates:
+            try:
+                mtime = f.stat().st_mtime
+            except OSError:
+                continue
+            if mtime <= cutoff:
+                try:
+                    shown = str(f.relative_to(worktree))
+                except ValueError:
+                    shown = rel
+                stale.append((mtime, shown))
+    return stale
+
+
+def scan_worktree_spec_drift(
+    repo_dir: Path,
+    *,
+    age_days: int = _WORKTREE_DRIFT_AGE_DAYS,
+    now: float | None = None,
+) -> list[WorktreeSpecDrift]:
+    """Week-old uncommitted spec content across the repo's worktrees.
+
+    Scans every worktree ``git worktree list`` reports for ``repo_dir``
+    (the main checkout included), looking at dirty/untracked files under
+    ``specs/`` and ``docs/specs/`` whose mtime is older than
+    ``age_days``. Best-effort and bounded: at most
+    ``_MAX_WORKTREES_SCANNED`` worktrees, one 2s-capped git call each,
+    and an overall ``_WORKTREE_SCAN_BUDGET_SECONDS`` wall-clock budget —
+    a slow or broken repo yields fewer findings, never an error.
+    """
+    started = time.monotonic()
+    cutoff = (now if now is not None else time.time()) - age_days * 86400
+    listing = _run_git(repo_dir, "worktree", "list", "--porcelain")
+    if not listing:
+        return []
+    worktrees = [
+        Path(line[len("worktree ") :].strip())
+        for line in listing.splitlines()
+        if line.startswith("worktree ")
+    ]
+    findings: list[WorktreeSpecDrift] = []
+    for worktree in worktrees[:_MAX_WORKTREES_SCANNED]:
+        if time.monotonic() - started > _WORKTREE_SCAN_BUDGET_SECONDS:
+            break
+        if not worktree.is_dir():
+            continue
+        try:
+            stale = _stale_spec_files(worktree, cutoff)
+        except Exception:  # noqa: BLE001 — one bad worktree must not kill the scan
+            continue
+        if not stale:
+            continue
+        stale.sort()
+        oldest_mtime, sample = stale[0]
+        findings.append(
+            WorktreeSpecDrift(
+                worktree=worktree.name,
+                since=time.strftime("%Y-%m-%d", time.localtime(oldest_mtime)),
+                sample=sample,
+                count=len(stale),
+            )
+        )
+    return findings
+
+
 def git_state(cwd: Path) -> GitState:
     """Return branch, last commit, and uncommitted files for ``cwd``.
 

.claude/hooks/spec_audit.py

@@ -36,6 +36,7 @@
 from __future__ import annotations
 
 import json
+import os
 import subprocess
 import sys
 import time
@@ -81,9 +82,20 @@
 
 # ── 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
+# The first full-workspace sweep (94 specs) capped out at 30 even after
+# the #1314 precision fix, so the weekly CI raises it via env; local
+# runs keep the conservative default.
+def _max_gh_calls() -> int:
+    try:
+        return int(os.environ.get("ATTUNE_SPEC_AUDIT_MAX_GH_CALLS", "30"))
+    except ValueError:
+        return 30
+
+
+_MAX_GH_CALLS = _max_gh_calls()
 _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
@@ -126,17 +138,30 @@ class _PrResolver:
       §2 — never blocks).
     """
 
-    def __init__(self, max_calls: int = _MAX_GH_CALLS) -> None:
+    def __init__(self, max_calls: int | None = None) -> None:
         self.calls = 0
+        # Resolved at construction (not def-time) so tests can tune the
+        # module constant; the explicit param was silently ignored
+        # pre-#1314.
+        self.max_calls = _MAX_GH_CALLS if max_calls is None else max_calls
         self.dead = False
+        self.failures = 0  # transient gh errors (timeout / non-zero exit)
+        self.capped = False  # ran out of the per-run call budget
         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:
+        if self.dead:
+            return False
+        if self.calls >= self.max_calls:
+            # Surfaced in the JSON payload — a capped run means "some
+            # refs unchecked", which reads very differently from clean
+            # (attune-ai#1314: silent gaps made consecutive runs
+            # disagree).
+            self.capped = True
             return False
         self.calls += 1
         try:
@@ -146,10 +171,15 @@ def merged(self, ref: PrRef, spec_dir: Path, layer: str) -> bool:
             # (offline degradation; the deliverable signal still stands).
             self.dead = True
             return False
+        if verdict is None:
+            # Transient failure: report it, and do NOT cache — a retry
+            # in the same run (other phase file) may still succeed.
+            self.failures += 1
+            return False
         self._cache[key] = verdict
         return verdict
 
-    def _check(self, ref: PrRef, spec_dir: Path) -> bool:
+    def _check(self, ref: PrRef, spec_dir: Path) -> bool | None:
         # 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
@@ -162,14 +192,19 @@ def _check(self, ref: PrRef, spec_dir: Path) -> bool:
                 ["pr", "view", str(ref.number), "--json", "state"],
                 cwd=spec_dir,
             )
-        if proc is None or proc.returncode != 0:
-            return False
+        if proc is None:
+            return None  # timeout / OSError — transient, retryable
+        if proc.returncode != 0:
+            # Non-zero is definitive for cross-repo (404 = not a PR) but
+            # can be transient for local gh auth hiccups; treating it as
+            # a failure keeps the run honest either way.
+            return None
         try:
             data = json.loads(proc.stdout)
         except (json.JSONDecodeError, ValueError):
-            return False
+            return None
         if not isinstance(data, dict):
-            return False
+            return None
         if ref.repo:
             return bool(data.get("merged"))
         return data.get("state") == "MERGED"
@@ -274,10 +309,13 @@ def audit_specs(
         checked = pr_links and resolver is not None and in_flight
         if checked:
             try:
+                # Only confident citations drive drift (attune-ai#1314):
+                # explicit PR #N / pull-URLs / shipping-context bare refs.
+                # Ordinal prose ("failure-shape #1") never reaches gh.
                 merged_prs = tuple(
                     _pr_label(ref, spec.layer)
                     for ref in _collect_refs(spec.path)
-                    if resolver.merged(ref, spec.path, spec.layer)
+                    if ref.explicit and resolver.merged(ref, spec.path, spec.layer)
                 )
             except Exception:  # noqa: BLE001 — one bad spec must not abort the audit
                 merged_prs = ()
@@ -337,7 +375,7 @@ def write_drift_cache(results: list[AuditResult], roots: list[Path]) -> list[Pat
     return written
 
 
-def format_json(results: list[AuditResult]) -> str:
+def format_json(results: list[AuditResult], resolver: _PrResolver | None = None) -> str:
     """Machine-readable audit payload for the tracking-issue upsert."""
     drifted = sorted(r.cache_key for r in results if r.drifted)
     payload = {
@@ -347,6 +385,14 @@ def format_json(results: list[AuditResult]) -> str:
             "drifted": len(drifted),
             "suspected_stale": sum(1 for r in results if r.staleness == "suspected-stale"),
         },
+        # attune-ai#1314: a capped or failure-ridden run means "refs went
+        # unchecked" — consumers (the weekly issue upsert, humans) must
+        # be able to tell that apart from a genuinely clean sweep.
+        "resolver": {
+            "calls": resolver.calls if resolver else 0,
+            "failures": resolver.failures if resolver else 0,
+            "capped": bool(resolver.capped) if resolver else False,
+        },
         "drifted": drifted,
         "specs": {
             r.cache_key: {
@@ -454,7 +500,14 @@ def main(argv: list[str] | None = None) -> int:
             # 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))
+        print(format_json(results, resolver) if as_json else format_report(results))
+        if resolver and (resolver.failures or resolver.capped):
+            print(
+                f"note: PR resolution degraded — {resolver.failures} lookup "
+                f"failure(s){', call budget hit' if resolver.capped else ''}; "
+                "unchecked refs count as not-drifted this run",
+                file=sys.stderr,
+            )
         if strict and any(r.drifted or r.staleness == "suspected-stale" for r in results):
             return 1
         return 0

.claude/hooks/spec_orient.py

@@ -45,9 +45,11 @@
 
 from _state import (  # noqa: E402 — sys.path bootstrap above
     SpecInfo,
+    WorktreeSpecDrift,
     discover_specs,
     prune_stale_sentinels,
     read_drift_cache,
+    scan_worktree_spec_drift,
     workspace_roots,
 )
 
@@ -148,6 +150,26 @@ def format_orientation(
     return "\n".join(lines)
 
 
+def format_worktree_drift(findings: list[WorktreeSpecDrift]) -> str:
+    """Alarm block for week-old uncommitted spec content in worktrees.
+
+    Task 10 (spec-status-integrity): this hook reads local files, so a
+    spec that lives only in a worktree still renders as "approved" in
+    the orientation above — these lines expose that instead of masking
+    it. Empty string when there is nothing to report.
+    """
+    if not findings:
+        return ""
+    lines = []
+    for f in findings:
+        more = f" (+{f.count - 1} more)" if f.count > 1 else ""
+        lines.append(
+            f"⚠ approved-looking spec content uncommitted in worktree "
+            f"{f.worktree} since {f.since}: {f.sample}{more} — commit or sweep it"
+        )
+    return "\n".join(lines)
+
+
 def render_spec_pin(spec: SpecInfo, char_budget: int = _POST_COMPACT_CHAR_BUDGET) -> str:
     """Render a spec body for post-compact context restoration.
 
@@ -211,6 +233,15 @@ def main() -> int:
             orient = format_orientation(specs, drift_cache=drift_cache, annotate=annotate)
             if orient:
                 print(orient)
+            if annotate:
+                # Task 10 — worktree-drift alarm. Same kill switch as the
+                # other audit annotations; bounded + best-effort inside.
+                try:
+                    alarm = format_worktree_drift(scan_worktree_spec_drift(cwd))
+                except Exception:  # noqa: BLE001 — never break SessionStart
+                    alarm = ""
+                if alarm:
+                    print(alarm)
         return 0
     except Exception:  # noqa: BLE001 — hook must never crash a session
         # Log the full traceback to stderr so plugin authors can