#!/usr/bin/env python3 """nightwatch.py — Phase 0 read-only maintenance CLI (Phase 1: ledger, normalization, ingest). Phase 0 is HAND-RUN and READ-ONLY. The ONLY thing this tool may write is an append-only event ledger (creating only its parent dir). It NEVER writes source, GitHub, Linear, branches, PRs, launchd/cron, report files, or temp schema files. Subcommands: ingest — scan codex-audit findings JSON, the hygiene queue, and (optionally) ruff, deriving stable identity for each observation and appending an "observed" event per finding to the ledger. verify — NotImplementedError (later phase). report — NotImplementedError (later phase). """ from __future__ import annotations import argparse import hashlib import json import logging import re import shlex import subprocess import sys from datetime import datetime, timezone from pathlib import Path # Bootstrap the repo root onto sys.path so the `recoil.`-prefixed import below # resolves when this CLI is run as a script (`python3 nightwatch.py`) or via its # shebang — not only via `python -m recoil.pipeline.tools.nightwatch`. This file # is recoil/pipeline/tools/nightwatch.py, so parents[3] is the repo root (the dir # that contains the recoil/ package). _REPO_ROOT = Path(__file__).resolve().parents[3] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from recoil.pipeline.tools.consult import run_codex_consultation # noqa: E402 from recoil.pipeline.tools.linear_queue import ( # noqa: E402 derive_claim_signature, # noqa: F401 # re-exported for nightwatch namespace (tested) effective_claim_signature, ) # --------------------------------------------------------------------------- # Constants (exact derivations — see spec). # This file is recoil/pipeline/tools/nightwatch.py: # parents[0]=tools, [1]=pipeline, [2]=recoil, [3]=repo root (CLAUDE_PROJECTS). # --------------------------------------------------------------------------- DEFAULT_REPO_ROOT = Path(__file__).resolve().parents[3] DEFAULT_LEDGER = Path.home() / "Dropbox/Claude_Config/maintenance/nightwatch/events.jsonl" DEFAULT_AUDIT_DIR = DEFAULT_REPO_ROOT / "overnight-reviews/audit" DEFAULT_HYGIENE_QUEUE = Path.home() / "Dropbox/Claude_Config/maintenance/hygiene-queue.md" SCHEMA_VERSION = 1 DEFAULT_VERIFY_SCHEMA = Path(__file__).resolve().parent / "nightwatch_verify_schema.json" LOGGER = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Primitive helpers. # --------------------------------------------------------------------------- def utc_now_iso() -> str: """ISO-8601 UTC with a literal Z suffix (not +00:00).""" return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def sha256_text(text: str) -> str: return "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest() def stable_json_digest(value: object) -> str: return sha256_text(json.dumps(value, sort_keys=True, separators=(",", ":"))) def current_head(repo_root: Path) -> str: """`git -C rev-parse --short HEAD`, or "unknown" on any failure.""" try: result = subprocess.run( ["git", "-C", str(repo_root), "rev-parse", "--short", "HEAD"], capture_output=True, text=True, check=False, ) except (FileNotFoundError, OSError): return "unknown" if result.returncode != 0: return "unknown" head = result.stdout.strip() return head or "unknown" # --------------------------------------------------------------------------- # Ledger I/O — APPEND-ONLY. Never overwrite, truncate, or atomically replace. # --------------------------------------------------------------------------- def append_event(ledger: Path, event: dict) -> None: ledger.parent.mkdir(parents=True, exist_ok=True) with open(ledger, "a", encoding="utf-8") as fh: fh.write(json.dumps(event, sort_keys=True, separators=(",", ":")) + "\n") def load_events(ledger: Path) -> list[dict]: if not ledger.exists(): return [] events: list[dict] = [] with open(ledger, "r", encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue events.append(json.loads(line)) return events # --------------------------------------------------------------------------- # Normalization + identity derivation. # --------------------------------------------------------------------------- def normalize_text(value: str) -> str: """Collapse internal whitespace runs to a single space; strip ends. Preserves code identifiers and quoted spans as-is (we only collapse whitespace, never lowercase or strip punctuation). None/empty -> "". """ if not value: return "" return re.sub(r"\s+", " ", value).strip() def derive_claim_fingerprint(title: str, evidence: str, recommendation: str) -> str: """Stable fingerprint of the substantive claim — case preserved (only whitespace normalized).""" parts = [ normalize_text(title), normalize_text(evidence), normalize_text(recommendation), ] return sha256_text("|".join(parts)) def derive_finding_key( category: str, file: str | None, claim_signature: str, ) -> str: """Stable, title-independent identity for a file-based finding.""" parts = [ (category or "").lower(), file or "", claim_signature, ] return sha256_text("|".join(parts)) def derive_observation_id( run_id: str | None, head_sha: str | None, source_report: str | None, source_record_digest: str | None, evidence_digest: str | None, ) -> str: """Per-observation identity (changes across runs / reports / evidence).""" parts = [ run_id or "", head_sha or "", source_report or "", source_record_digest or "", evidence_digest or "", ] return sha256_text("|".join(parts)) # Spans: backtick-quoted, double-quoted, single-quoted. _BACKTICK_RE = re.compile(r"`([^`]+)`") _DQUOTE_RE = re.compile(r'"([^"]+)"') _SQUOTE_RE = re.compile(r"'([^']+)'") _SYMBOL_DEF_RE = re.compile(r"^\s*(def |class |async def )") def _candidate_spans(evidence: str) -> list[str]: spans: list[str] = [] for rx in (_BACKTICK_RE, _DQUOTE_RE, _SQUOTE_RE): spans.extend(rx.findall(evidence or "")) return spans def extract_normalized_anchor( file_path: Path | None, evidence: str, advisory_line: int | None, ) -> tuple[str, list[str]]: """Return (anchor, blocked_reasons). 1. Prefer the LONGEST quoted candidate span from `evidence` that — after whitespace-normalization — appears in the current file's normalized text. 2. Else, if the file exists and advisory_line is set, scan a 20-line window around it for the nearest def/class/symbol-definition line. 3. Else, return a normalized excerpt of evidence and ["anchor_not_found"]. All file reads are guarded. If file_path is given but missing, add "missing_file" to blocked_reasons and fall through to step 3. """ blocked: list[str] = [] file_text: str | None = None file_lines: list[str] | None = None file_missing = file_path is not None and not file_path.exists() if file_missing: blocked.append("missing_file") if file_path is not None and file_path.exists(): try: file_text = file_path.read_text(encoding="utf-8", errors="replace") file_lines = file_text.splitlines() except (OSError, UnicodeError): file_text = None file_lines = None # Step 1: longest quoted span that appears in the file's normalized text. if file_text is not None: normalized_file = normalize_text(file_text) candidates = _candidate_spans(evidence) # Longest candidate (by normalized length) first. best: str | None = None for cand in candidates: norm = normalize_text(cand) if not norm: continue if norm in normalized_file: if best is None or len(norm) > len(best): best = norm if best is not None: return best, [] # Step 2: nearest symbol definition in a 20-line window around advisory_line. if file_lines is not None and advisory_line is not None: idx = advisory_line - 1 # 1-based -> 0-based lo = max(0, idx - 20) hi = min(len(file_lines), idx + 21) nearest_line: str | None = None nearest_dist: int | None = None for i in range(lo, hi): if _SYMBOL_DEF_RE.match(file_lines[i]): dist = abs(i - idx) if nearest_dist is None or dist < nearest_dist: nearest_dist = dist nearest_line = file_lines[i] if nearest_line is not None: return normalize_text(nearest_line), [] # Step 3: fallback excerpt. blocked.append("anchor_not_found") return normalize_text(evidence)[:200], blocked # --------------------------------------------------------------------------- # Event envelope construction. # --------------------------------------------------------------------------- def _build_event( *, event_type: str, run_id: str, repo_root: Path, head_sha: str, finding_key: str, observation_id: str, judgment_source: str, payload: dict, ) -> dict: """Assemble an envelope, computing event_id over the body (sans event_id + event_ts). Single home for the envelope + event_id digest scheme shared by observed and verified events (SSOT — a future envelope change touches exactly one place). """ event = { "schema_version": SCHEMA_VERSION, "event_ts": utc_now_iso(), "event_type": event_type, "run_id": run_id, "repo_root": str(repo_root), "head_sha": head_sha, "finding_key": finding_key, "observation_id": observation_id, "judgment_source": judgment_source, "payload": payload, } digest_input = {k: v for k, v in event.items() if k not in ("event_id", "event_ts")} event["event_id"] = stable_json_digest(digest_input) return event def _build_observed_event( *, run_id: str, repo_root: Path, head_sha: str, finding_key: str, observation_id: str, payload: dict, ) -> dict: """Observed envelope (judgment_source is always nightwatch_ingest).""" return _build_event( event_type="observed", run_id=run_id, repo_root=repo_root, head_sha=head_sha, finding_key=finding_key, observation_id=observation_id, judgment_source="nightwatch_ingest", payload=payload, ) # --------------------------------------------------------------------------- # Risk classification (codex_audit categories). # --------------------------------------------------------------------------- def _classify_codex_category(category: str) -> tuple[str, str]: """Return (risk_class, classification) for a codex_audit category.""" cat = (category or "").lower() if cat in ("ssot", "architectural-law", "drift", "dead-code"): return "escalation", "would_escalate" if cat == "bug": return "escalation", "would_escalate" if cat == "efficiency": # Phase 0: default to report; do not over-engineer "clearly mechanical". return "report", "report" # Unknown category — treat conservatively as escalation. return "escalation", "would_escalate" # --------------------------------------------------------------------------- # Ingest: codex_audit findings JSON. # --------------------------------------------------------------------------- def ingest_codex_audit(audit_json_path: Path, run_id: str, repo_root: Path) -> list[dict]: """Parse a codex_audit findings JSON file into observed events (not appended here).""" try: raw = json.loads(audit_json_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return [] findings = raw.get("findings") if isinstance(raw, dict) else None if not isinstance(findings, list): return [] root_head = raw.get("head_sha") if isinstance(raw, dict) else None head_sha = root_head if root_head else current_head(repo_root) source_report = str(audit_json_path) events: list[dict] = [] for finding in findings: if not isinstance(finding, dict): continue category = finding.get("category") or "" law_or_rule = finding.get("law_or_rule") file_rel = finding.get("file") # repo-relative already line = finding.get("line") title = finding.get("title") or "" evidence = finding.get("evidence") or "" recommendation = finding.get("recommendation") or "" line_val = line if isinstance(line, int) else None blocked_reason: list[str] = [] # Resolve file against repo_root ONLY to read it for anchor extraction. abs_file: Path | None = None if file_rel: abs_file = (repo_root / file_rel) # Path-traversal guard: ensure the resolved path stays under repo_root. try: abs_file.resolve().relative_to(repo_root.resolve()) except ValueError: blocked_reason.append("path_traversal") abs_file = None normalized_anchor, anchor_blocked = extract_normalized_anchor( abs_file, evidence, line_val ) blocked_reason.extend(anchor_blocked) claim_fingerprint = derive_claim_fingerprint(title, evidence, recommendation) claim_signature = effective_claim_signature(evidence, normalized_anchor) finding_key = derive_finding_key(category, file_rel, claim_signature) source_record_digest = stable_json_digest(finding) evidence_digest = stable_json_digest( {"title": title, "evidence": evidence, "recommendation": recommendation} ) observation_id = derive_observation_id( run_id, head_sha, source_report, source_record_digest, evidence_digest ) risk_class, classification = _classify_codex_category(category) payload = { "source_type": "codex_audit", "source_report": source_report, "source_record_digest": source_record_digest, "category": category.lower(), "risk_class": risk_class, "classification": classification, "law_or_rule": law_or_rule, "file": file_rel, "line": line_val, "line_is_advisory": True, "normalized_anchor": normalized_anchor, "claim_fingerprint": claim_fingerprint, "title": title, "evidence": evidence, "recommendation": recommendation, "blocked_reason": blocked_reason, "shadow_gate": None, } events.append( _build_observed_event( run_id=run_id, repo_root=repo_root, head_sha=head_sha, finding_key=finding_key, observation_id=observation_id, payload=payload, ) ) return events # --------------------------------------------------------------------------- # Ingest: hygiene-queue.md open items. # --------------------------------------------------------------------------- _FILELINE_RE = re.compile(r"^.+:\d+(,\d+)*$") def _parse_hygiene_line(raw_line: str) -> dict | None: """Parse one `- DATE | file:line | issue | why` open-item line, or None if not a match.""" if not raw_line.startswith("- "): return None # Need at least 3 separators -> at least 4 fields. if raw_line.count("|") < 3: return None fields = [f.strip() for f in raw_line.split("|", 3)] # Field 0 = date (strip leading "- "); 1 = file:line; 2 = issue; 3 = why. date = fields[0][2:].strip() if fields[0].startswith("- ") else fields[0] file_line = fields[1] issue = fields[2] if len(fields) > 2 else "" why = fields[3] if len(fields) > 3 else "" if not _FILELINE_RE.match(file_line): return None file_part, _, line_spec = file_line.rpartition(":") if not file_part: return None # First integer in the line-spec is the advisory line. line_val: int | None = None first_token = line_spec.split(",")[0].strip() try: line_val = int(first_token) except ValueError: line_val = None return { "date": date, "file": file_part, "line": line_val, "issue": issue, "why": why, } def ingest_hygiene_queue(hygiene_path: Path, run_id: str, repo_root: Path) -> list[dict]: """Parse the hygiene queue's `## Open` items into observed events (not appended here).""" try: text = hygiene_path.read_text(encoding="utf-8") except OSError: return [] head_sha = current_head(repo_root) source_report = str(hygiene_path) events: list[dict] = [] in_open = False for raw in text.splitlines(): stripped = raw.strip() if stripped.startswith("## "): # Only items under the `## Open` heading are live. Any other `## ` # section (Archived/Resolved/Done/…) is closed and must be skipped, # so resolved items aren't re-ingested as fresh observations. in_open = stripped[3:].strip().lower().startswith("open") continue if not in_open: continue parsed = _parse_hygiene_line(raw.rstrip("\n")) if parsed is None: continue file_rel = parsed["file"] line_val = parsed["line"] title = parsed["issue"] evidence = parsed["issue"] recommendation = parsed["why"] blocked_reason: list[str] = [] abs_file: Path | None = None if file_rel: abs_file = repo_root / file_rel try: abs_file.resolve().relative_to(repo_root.resolve()) except ValueError: blocked_reason.append("path_traversal") abs_file = None normalized_anchor, anchor_blocked = extract_normalized_anchor( abs_file, evidence, line_val ) blocked_reason.extend(anchor_blocked) claim_fingerprint = derive_claim_fingerprint(title, evidence, recommendation) claim_signature = effective_claim_signature(evidence, normalized_anchor) finding_key = derive_finding_key("hygiene", file_rel, claim_signature) source_record_digest = stable_json_digest({"raw": raw}) evidence_digest = stable_json_digest( {"title": title, "evidence": evidence, "recommendation": recommendation} ) observation_id = derive_observation_id( run_id, head_sha, source_report, source_record_digest, evidence_digest ) payload = { "source_type": "hygiene_queue", "source_report": source_report, "source_record_digest": source_record_digest, "category": "hygiene", "risk_class": "mechanical", "classification": "would_fix", "law_or_rule": None, "file": file_rel, "line": line_val, "line_is_advisory": True, "normalized_anchor": normalized_anchor, "claim_fingerprint": claim_fingerprint, "title": title, "evidence": evidence, "recommendation": recommendation, "blocked_reason": blocked_reason, "shadow_gate": None, } events.append( _build_observed_event( run_id=run_id, repo_root=repo_root, head_sha=head_sha, finding_key=finding_key, observation_id=observation_id, payload=payload, ) ) return events # --------------------------------------------------------------------------- # Ingest: ruff scanner. # --------------------------------------------------------------------------- def ingest_ruff(repo_root: Path, run_id: str) -> list[dict]: """Run `ruff check --output-format=json` and turn each item into an observed event. Nonzero exit is NORMAL (ruff exits nonzero when it finds issues) — parse stdout JSON anyway. If the binary is missing or stdout is unparseable, return []. """ try: result = subprocess.run( ["ruff", "check", "--output-format=json"], cwd=str(repo_root), capture_output=True, text=True, check=False, ) except (FileNotFoundError, OSError): return [] try: items = json.loads(result.stdout) except (json.JSONDecodeError, TypeError): return [] if not isinstance(items, list): return [] head_sha = current_head(repo_root) source_report = "ruff check --output-format=json" events: list[dict] = [] for item in items: if not isinstance(item, dict): continue code = item.get("code") or "" message = item.get("message") or "" filename = item.get("filename") or "" location = item.get("location") or {} line_val = location.get("row") if isinstance(location, dict) else None if not isinstance(line_val, int): line_val = None col_val = location.get("column") if isinstance(location, dict) else None # Store a repo-relative path if the filename is under repo_root. blocked_reason: list[str] = [] file_rel: str | None = filename or None abs_file: Path | None = None if filename: # Anchor to repo_root (matches codex/hygiene ingest). For an absolute # ruff path, `repo_root / abs` == abs; for a relative one it resolves # under repo_root instead of the process cwd. candidate = repo_root / filename try: file_rel = str(candidate.resolve().relative_to(repo_root.resolve())) abs_file = candidate except ValueError: # Path escapes repo_root — do NOT read it (consistent with the # codex/hygiene ingest traversal guard; never read outside the repo). file_rel = filename abs_file = None blocked_reason.append("path_traversal") title = f"ruff {code}".strip() evidence = message recommendation = "ruff offers an autofix" if isinstance(item.get("fix"), dict) else "" normalized_anchor, anchor_blocked = extract_normalized_anchor( abs_file, evidence, line_val ) blocked_reason.extend(anchor_blocked) claim_fingerprint = derive_claim_fingerprint(title, evidence, recommendation) claim_signature = "|".join([ str(code or ""), str(line_val if line_val is not None else ""), str(col_val if isinstance(col_val, int) else ""), normalize_text(message), normalized_anchor, ]) finding_key = derive_finding_key("scanner", file_rel, claim_signature) source_record_digest = stable_json_digest(item) evidence_digest = stable_json_digest( {"title": title, "evidence": evidence, "recommendation": recommendation} ) observation_id = derive_observation_id( run_id, head_sha, source_report, source_record_digest, evidence_digest ) payload = { "source_type": "scanner", "source_report": source_report, "source_record_digest": source_record_digest, "category": "scanner", "risk_class": "mechanical", "classification": "would_fix", "law_or_rule": code or None, "file": file_rel, "line": line_val, "line_is_advisory": True, "normalized_anchor": normalized_anchor, "claim_fingerprint": claim_fingerprint, "title": title, "evidence": evidence, "recommendation": recommendation, "blocked_reason": blocked_reason, "shadow_gate": None, } events.append( _build_observed_event( run_id=run_id, repo_root=repo_root, head_sha=head_sha, finding_key=finding_key, observation_id=observation_id, payload=payload, ) ) return events # --------------------------------------------------------------------------- # Audit JSON selection. # --------------------------------------------------------------------------- # Findings files are named findings---.json. _AUDIT_TS_RE = re.compile(r"(\d{8})-(\d{6})\.json$") def _audit_sort_key(path: Path) -> tuple: """Newest-run sort key: the trailing YYYYMMDD-HHMMSS timestamp (NOT the whole filename, which would sort by subsystem slug first). Timestamped files rank above un-timestamped ones; the filename is a deterministic final tiebreaker.""" m = _AUDIT_TS_RE.search(path.name) stamp = m.group(1) + m.group(2) if m else "" return (1 if m else 0, stamp, path.name) def _latest_findings_json(audit_dir: Path) -> Path | None: """Return the findings JSON for the NEWEST audit RUN (by filename timestamp), not the alphabetically-last subsystem — codex_audit writes one file per subsystem, so plain filename sort would pick the wrong (older) run.""" if not audit_dir.exists(): return None candidates = list(audit_dir.glob("findings-*.json")) if not candidates: return None return max(candidates, key=_audit_sort_key) # --------------------------------------------------------------------------- # Subcommand entry points. # --------------------------------------------------------------------------- def cmd_ingest(args: argparse.Namespace) -> int: repo_root = Path(args.repo_root) ledger = Path(args.ledger) run_id = "nightwatch-" + datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") all_events: list[dict] = [] # 1. codex audit JSON. audit_json: Path | None = None if args.audit_json: candidate = Path(args.audit_json) if candidate.exists(): audit_json = candidate else: print(f"[nightwatch] audit-json not found: {candidate} — skipping codex audit", file=sys.stderr) else: audit_json = _latest_findings_json(Path(args.audit_dir)) if audit_json is None: print(f"[nightwatch] no findings-*.json under {args.audit_dir} — skipping codex audit", file=sys.stderr) if audit_json is not None: codex_events = ingest_codex_audit(audit_json, run_id, repo_root) all_events.extend(codex_events) print(f"[nightwatch] codex_audit: {len(codex_events)} observations from {audit_json}", file=sys.stderr) # 2. hygiene queue. hygiene_path = Path(args.hygiene_queue) if hygiene_path.exists(): hygiene_events = ingest_hygiene_queue(hygiene_path, run_id, repo_root) all_events.extend(hygiene_events) print(f"[nightwatch] hygiene_queue: {len(hygiene_events)} observations from {hygiene_path}", file=sys.stderr) else: print(f"[nightwatch] hygiene queue not found: {hygiene_path} — skipping", file=sys.stderr) # 3. optional ruff. if args.include_ruff: ruff_events = ingest_ruff(repo_root, run_id) all_events.extend(ruff_events) print(f"[nightwatch] ruff: {len(ruff_events)} observations", file=sys.stderr) for event in all_events: append_event(ledger, event) print( f"[nightwatch] run_id={run_id} appended {len(all_events)} events to {ledger}", file=sys.stderr, ) return 0 # --------------------------------------------------------------------------- # Phase 3: Verify Gate. # --------------------------------------------------------------------------- def load_unverified_observations(events: list[dict], limit: int | None) -> list[dict]: """Return observed events whose observation_id has no verified event yet. De-dups to the latest observed event per observation_id (ledger order). limit None or <= 0 means no cap; otherwise return at most `limit`. """ verified_ids = { e.get("observation_id") for e in events if e.get("event_type") == "verified" } latest_by_obs: dict[str, dict] = {} order: list[str] = [] for e in events: if e.get("event_type") != "observed": continue obs_id = e.get("observation_id") if obs_id is None: continue if obs_id in verified_ids: continue if obs_id not in latest_by_obs: order.append(obs_id) latest_by_obs[obs_id] = e # latest observed wins result = [latest_by_obs[obs_id] for obs_id in order] if limit is None or limit <= 0: return result return result[:limit] def replay_observation(event: dict, repo_root: Path) -> dict: """Deterministic replay of an observed event against current code. Returns {"deterministic_status", "reason", "file_resolved", "anchor_present"}. """ payload = event.get("payload") or {} file_rel = payload.get("file") normalized_anchor = payload.get("normalized_anchor") or "" anchor_not_found = "anchor_not_found" in (payload.get("blocked_reason") or []) # Path traversal is a hard security reject — check it FIRST when a file is # named (a traversal finding also carries anchor_not_found, so the abstain # branch below must not be allowed to mask the rejection). abs_file: Path | None = None if file_rel: abs_file = repo_root / file_rel try: abs_file.resolve().relative_to(repo_root.resolve()) except ValueError: return { "deterministic_status": "rejected", "reason": "path outside repo", "file_resolved": False, "anchor_present": False, } # An anchor that was never groundable at ingest (a fallback evidence excerpt) # must replay to abstain — "we can't ground this", not "the code changed". # This DOMINATES no-file / file-missing / drift (spec: anchor_not_found -> # abstain, not confirmed), but ranks below the traversal rejection above. if anchor_not_found: return { "deterministic_status": "abstain", "reason": "anchor not groundable at ingest (anchor_not_found)", "file_resolved": bool(file_rel), "anchor_present": False, } if not file_rel: # A finding with no file cannot be deterministically grounded against # current code -> abstain (spec: abstain when the claim can't be # grounded). Never route an ungroundable finding to the model. return { "deterministic_status": "abstain", "reason": "no file to ground against deterministically", "file_resolved": False, "anchor_present": False, } if not abs_file.exists(): return { "deterministic_status": "stale", "reason": "file missing", "file_resolved": False, "anchor_present": False, } try: file_text = abs_file.read_text(encoding="utf-8", errors="replace") except OSError: return { "deterministic_status": "stale", "reason": "file unreadable", "file_resolved": True, "anchor_present": False, } if normalized_anchor: normalized_file = normalize_text(file_text) if normalize_text(normalized_anchor) not in normalized_file: return { "deterministic_status": "stale", "reason": "anchor drift", "file_resolved": True, "anchor_present": False, } return { "deterministic_status": "current", "reason": "anchor present", "file_resolved": True, "anchor_present": True, } # No anchor to check — file exists; let the model judge. return { "deterministic_status": "current", "reason": "no anchor to replay", "file_resolved": True, "anchor_present": False, } def check_duplicate_pr(event: dict, repo_root: Path) -> str: """Read-only, optional open-PR duplicate check. Any failure -> "unknown".""" try: import shutil if shutil.which("gh") is None: return "unknown" # Search by the precise finding_key only. A title search would risk # false "duplicate" hits that suppress a real finding from the report; # finding_key is exact and forward-compatible (matches PRs that embed it). query = event.get("finding_key") or "" if not query: return "unknown" result = subprocess.run( [ "gh", "pr", "list", "--search", str(query), "--state", "open", "--json", "number,title,url", ], cwd=str(repo_root), capture_output=True, text=True, check=False, ) if result.returncode != 0: return "unknown" prs = json.loads(result.stdout) if isinstance(prs, list) and prs: return "open_pr_duplicate" return "none" except Exception: return "unknown" def build_verify_prompt(observed_event: dict, replay: dict) -> str: """Build a strict verification prompt asking only whether the finding is real.""" payload = observed_event.get("payload") or {} file_rel = payload.get("file") line = payload.get("line") title = payload.get("title") or "" evidence = payload.get("evidence") or "" recommendation = payload.get("recommendation") or "" category = payload.get("category") or "" normalized_anchor = payload.get("normalized_anchor") or "" replay_status = replay.get("deterministic_status") return ( "Verify whether the following maintenance finding is REAL against the " "CURRENT code and the cited evidence. Judge ONLY truth, not desirability.\n\n" "You are running READ-ONLY against the live checkout. You MAY and SHOULD " "open the cited file and any directly-related file to confirm or reject " "this finding against the CURRENT code. Ground your verdict in what the " "file actually contains — do not abstain merely because the evidence " "string is not self-contained; open the file.\n\n" f"File: {file_rel}\n" f"Line (ADVISORY ONLY — line drift is not disqualifying): {line}\n" f"Category: {category}\n" f"Title: {title}\n" f"Evidence: {evidence}\n" f"Recommendation (context only — do NOT act on it): {recommendation}\n" f"Normalized anchor: {normalized_anchor}\n" f"Deterministic replay status: {replay_status}\n\n" "Decide:\n" "- 'confirmed' — the finding is real and grounded in the current code/evidence.\n" "- 'rejected' — the finding is NOT real (already fixed, wrong, or contradicted by the code).\n" "- 'abstain' — the claim genuinely cannot be grounded after reading the file. " "You MUST abstain rather than guess, but abstain ONLY when the claim genuinely " "cannot be grounded after reading the file.\n\n" "Constraints on your OUTPUT (these do not restrict reading): do not propose fixes, " "do not write issue text, do not plan PRs or Linear actions.\n" "- Return ONLY the JSON object matching the provided schema: " "{\"verdict\", \"reason\", \"confidence\"}.\n" ) def parse_model_verdict(text: str) -> dict: """Parse a model verdict JSON, defensively. Malformed -> abstain/low (never confirmed).""" def _abstain(detail: str) -> dict: return { "verdict": "abstain", "reason": f"model output unparseable: {detail}", "confidence": "low", } if not isinstance(text, str): return _abstain("non-string output") obj = None try: obj = json.loads(text) except (json.JSONDecodeError, TypeError): # Defensively extract the first {...} object from surrounding prose. start = text.find("{") end = text.rfind("}") if start != -1 and end != -1 and end > start: try: obj = json.loads(text[start:end + 1]) except (json.JSONDecodeError, TypeError): obj = None if not isinstance(obj, dict): return _abstain("not a JSON object") verdict = obj.get("verdict") reason = obj.get("reason") confidence = obj.get("confidence") if verdict not in ("confirmed", "rejected", "abstain"): return _abstain(f"bad verdict {verdict!r}") if not isinstance(reason, str): return _abstain("reason not a string") if confidence not in ("high", "medium", "low"): return _abstain(f"bad confidence {confidence!r}") return {"verdict": verdict, "reason": reason, "confidence": confidence} def _build_verified_event( *, run_id: str, repo_root: Path, head_sha: str, finding_key: str, observation_id: str, judgment_source: str, payload: dict, ) -> dict: """Verified envelope — delegates to the shared _build_event (SSOT).""" return _build_event( event_type="verified", run_id=run_id, repo_root=repo_root, head_sha=head_sha, finding_key=finding_key, observation_id=observation_id, judgment_source=judgment_source, payload=payload, ) def verify_observation( observed_event: dict, repo_root: Path, schema_path: str | Path, timeout: int = 3600, run_id: str | None = None, ) -> dict: """Verify ONE observation; return the verified event dict ready to append.""" replay = replay_observation(observed_event, repo_root) det_status = replay["deterministic_status"] model_confidence: str | None = None # The dup check spawns a `gh` subprocess; only the "current" branch can act # on it, so leave it "unknown" (not checked) for deterministically-decided # findings rather than burning a network round-trip per stale/rejected one. dup = "unknown" if det_status == "rejected": verify_status = "rejected" verify_reason = replay["reason"] judgment_source = "deterministic" elif det_status == "stale": verify_status = "stale" verify_reason = replay["reason"] judgment_source = "deterministic" elif det_status == "abstain": # anchor_not_found at ingest -> deterministic abstain, no model call. verify_status = "abstain" verify_reason = replay["reason"] judgment_source = "deterministic" else: # det_status == "current": check for an open-PR duplicate, else ask the model. dup = check_duplicate_pr(observed_event, repo_root) if dup == "open_pr_duplicate": verify_status = "duplicate" verify_reason = "open PR already addresses this finding" judgment_source = "deterministic" else: payload = observed_event.get("payload") or {} context = ( f"File: {payload.get('file')}\n" f"Normalized anchor: {payload.get('normalized_anchor') or ''}\n" f"Evidence: {payload.get('evidence') or ''}\n" f"Replay status: {replay['deterministic_status']} ({replay['reason']})\n" ) prompt = build_verify_prompt(observed_event, replay) text = run_codex_consultation( context=context, prompt=prompt, effort="high", timeout=timeout, output_schema=str(schema_path), cwd=repo_root, # ground the model against the verified checkout, not consult.py's ) verdict = parse_model_verdict(text) verify_status = verdict["verdict"] verify_reason = verdict["reason"] model_confidence = verdict["confidence"] judgment_source = "codex_consult" head_sha = current_head(repo_root) verify_payload = { "verify_status": verify_status, "verify_reason": verify_reason, "deterministic_status": det_status, "duplicate_status": dup, "model_confidence": model_confidence, } return _build_verified_event( run_id=run_id or observed_event.get("run_id") or "", repo_root=repo_root, head_sha=head_sha, finding_key=observed_event.get("finding_key"), observation_id=observed_event.get("observation_id"), judgment_source=judgment_source, payload=verify_payload, ) def _linear_queue_category(category: object) -> str: """Return the category spelling accepted by linear_queue.""" if isinstance(category, str) and category.lower() == "ssot": return "SSOT" return str(category or "") def enqueue_linear_confirmed_survivor( *, verified: dict, observed: dict, run_id: str, ) -> bool: """Best-effort enqueue of one confirmed Nightwatch survivor. Returns True only when an inbox event was written. Any queue failure is logged and suppressed so verify remains a Nightwatch-ledger operation. """ verified_payload = verified.get("payload") or {} verdict = verified_payload.get("verify_status") or verified_payload.get("verdict") if verdict != "confirmed": return False observed_payload = observed.get("payload") or {} p = dict(observed_payload) for key in ( "category", "file", "normalized_anchor", "law_or_rule", "title", "evidence", "recommendation", "subject_kind", "subject_id", "claim_fingerprint", "source_type", ): if key in verified_payload and verified_payload[key] is not None: p[key] = verified_payload[key] try: from recoil.pipeline.tools import linear_queue as lq file_rel = p.get("file") event_fields = { "source_type": "nightwatch", "session_id": run_id, "category": _linear_queue_category(p.get("category")), "file": file_rel, "normalized_anchor": p.get("normalized_anchor", ""), "law_or_rule": p.get("law_or_rule"), "title": p["title"], "evidence": p["evidence"], "recommendation": p["recommendation"], } if p.get("source_type") == "scanner" or p.get("category") == "scanner": event_fields["finding_key"] = observed.get("finding_key") if not file_rel: subject_kind = p.get("subject_kind") subject_id = p.get("subject_id") if subject_kind and subject_id: event_fields["subject_kind"] = subject_kind event_fields["subject_id"] = subject_id ev = lq.build_event(**event_fields) lq.atomic_write_inbox(ev) return True except Exception: LOGGER.exception( "failed to enqueue confirmed Nightwatch finding into Linear inbox: %s", verified.get("finding_key") or observed.get("finding_key"), ) return False def cmd_verify(args: argparse.Namespace) -> int: repo_root = Path(args.repo_root) ledger = Path(args.ledger) run_id = "nightwatch-" + datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") events = load_events(ledger) pending = load_unverified_observations(events, args.limit) appended = 0 for obs in pending: verified = verify_observation( obs, repo_root, schema_path=args.verify_schema, timeout=getattr(args, "timeout", 3600), run_id=run_id, ) append_event(ledger, verified) if getattr(args, "enqueue_linear", False): enqueue_linear_confirmed_survivor( verified=verified, observed=obs, run_id=run_id, ) appended += 1 print( f"[nightwatch] verify run_id={run_id}: {appended} of " f"{len(pending)} unverified observations verified -> {ledger}", file=sys.stderr, ) return 0 # --------------------------------------------------------------------------- # Phase 4: Report Projection (READ-ONLY — reads the ledger, prints to stdout). # This phase writes NOTHING: no report file, no ledger append. It only PRINTS # the append commands for a human to run by hand. # --------------------------------------------------------------------------- _RISK_RANK = {"escalation": 0, "mechanical": 1, "report": 2} _DIAGNOSTIC_STATUSES = {"rejected", "abstain", "stale", "duplicate", "unverified"} # Human adjudications that remove a finding from the confirmed report (the human # has said it's not real / not worth acting on now). "agree" / "wrong-classification" # do NOT suppress — the finding stays surfaced. _SUPPRESSING_ADJUDICATIONS = {"false-positive", "ignore-for-now"} def project_findings(events: list[dict], include_diagnostics: bool = False) -> list[dict]: """Project observed+verified events into one dict per finding_key. Design choice (per spec): this function ALWAYS returns the FULL list of projections (every finding_key), regardless of `include_diagnostics`. `render_report` is the single place that decides what to surface — confirmed only by default, plus a diagnostics section when include_diagnostics is True. `include_diagnostics` is accepted for signature symmetry but does not filter here. SSOT for the confirmed-vs-diagnostic split lives in render_report. Per finding_key: - latest observed event drives title/evidence/recommendation/source_type/ file/line/category/risk_class/classification/law_or_rule and the representative observation_id. - latest verified event (by ledger order) drives verify_status (+ reason, model_confidence, deterministic_status, duplicate_status). None -> the sentinel "unverified". - observation_count = distinct observation_ids among observed events. - first_seen/last_seen = min/max event_ts among observed events. Sorted by risk_class rank (escalation=0, mechanical=1, report=2, else=3), then last_seen DESC within rank. """ latest_observed: dict[str, dict] = {} observed_ids: dict[str, set] = {} collapsed_members: dict[str, dict[object, dict]] = {} seen_ts: dict[str, list[str]] = {} latest_verified: dict[str, dict] = {} latest_adjudication: dict[str, dict] = {} for e in events: et = e.get("event_type") fk = e.get("finding_key") if fk is None: continue if et == "observed": latest_observed[fk] = e # ledger order -> last wins observed_ids.setdefault(fk, set()).add(e.get("observation_id")) op = e.get("payload") or {} oid = e.get("observation_id") collapsed_members.setdefault(fk, {})[oid] = { "observation_id": oid, "title": op.get("title"), "evidence": op.get("evidence"), "normalized_anchor": op.get("normalized_anchor"), } seen_ts.setdefault(fk, []).append(e.get("event_ts")) elif et == "verified": latest_verified[fk] = e # last wins elif et == "human_adjudicated": latest_adjudication[fk] = e # last wins projections: list[dict] = [] for fk, obs in latest_observed.items(): op = obs.get("payload") or {} ver = latest_verified.get(fk) if ver is not None: vp = ver.get("payload") or {} verify_status = vp.get("verify_status") or "unverified" verify_reason = vp.get("verify_reason") model_confidence = vp.get("model_confidence") deterministic_status = vp.get("deterministic_status") duplicate_status = vp.get("duplicate_status") else: verify_status = "unverified" verify_reason = None model_confidence = None deterministic_status = None duplicate_status = None ts_values = [t for t in seen_ts.get(fk, []) if t is not None] first_seen = min(ts_values) if ts_values else None last_seen = max(ts_values) if ts_values else None projections.append( { "finding_key": fk, "observation_id": obs.get("observation_id"), "source_type": op.get("source_type"), "file": op.get("file"), "line": op.get("line"), "category": op.get("category"), "risk_class": op.get("risk_class"), "classification": op.get("classification"), "law_or_rule": op.get("law_or_rule"), "title": op.get("title"), "evidence": op.get("evidence"), "recommendation": op.get("recommendation"), "verify_status": verify_status, "verify_reason": verify_reason, "model_confidence": model_confidence, "deterministic_status": deterministic_status, "duplicate_status": duplicate_status, "observation_count": len(observed_ids.get(fk, set())), "distinct_observation_count": len(observed_ids.get(fk, set())), "collapsed_observations": sorted( collapsed_members.get(fk, {}).values(), key=lambda m: str(m.get("observation_id") or ""), ), "first_seen": first_seen, "last_seen": last_seen, "human_adjudication": ( (latest_adjudication[fk].get("payload") or {}).get("human_adjudication") if fk in latest_adjudication else None ), } ) projections.sort( key=lambda p: ( _RISK_RANK.get(p.get("risk_class"), 3), # last_seen DESC within rank: negate by reverse-sorting on the value. _neg_sort_key(p.get("last_seen")), ) ) return projections class _Reversed: """Wrapper to sort a value DESC within an otherwise-ascending sort key.""" __slots__ = ("value",) def __init__(self, value): self.value = value def __lt__(self, other): return self.value > other.value def _neg_sort_key(value): return _Reversed(value or "") def format_adjudication_command(ledger: Path, finding: dict, adjudication: str) -> str: """Return a shell-safe `printf ... >> ledger` command that appends ONE human_adjudicated event. PRINTED only — never executed by this tool. The embedded JSON is a static envelope (no computed event_id/event_ts): schema_version, event_type, finding_key, observation_id, judgment_source, and payload={"human_adjudication": adjudication, "note": null}. """ event = { "schema_version": SCHEMA_VERSION, "event_type": "human_adjudicated", "finding_key": finding["finding_key"], "observation_id": finding["observation_id"], "judgment_source": "human", "payload": {"human_adjudication": adjudication, "note": None}, } payload_json = json.dumps(event, separators=(",", ":")) return f"printf '%s\\n' {shlex.quote(payload_json)} >> {shlex.quote(str(ledger))}" def _render_finding_block(finding: dict, ledger: Path) -> str: lines = [ f"### [{finding.get('risk_class')}] {finding.get('title') or '(untitled)'}", "", f"- source_type: {finding.get('source_type')}", f"- file: {finding.get('file')}", f"- evidence: {finding.get('evidence')}", f"- recommendation: {finding.get('recommendation')}", f"- verify_reason: {finding.get('verify_reason')}", f"- observation_count: {finding.get('observation_count')}", f"- first_seen: {finding.get('first_seen')}", f"- last_seen: {finding.get('last_seen')}", f"- risk_class: {finding.get('risk_class')}", "", "```bash", ] for adjudication in ("agree", "wrong-classification", "false-positive", "ignore-for-now"): lines.append(format_adjudication_command(ledger, finding, adjudication)) lines.append("```") lines.append("") return "\n".join(lines) def _truncate_report_value(value: object, limit: int = 300) -> str: text = str(value or "") if len(text) <= limit: return text return text[: limit - 3] + "..." def _render_possible_collapse_section(projections: list[dict]) -> str: lines = [ "## Possible collapse (≥2 distinct observations share one finding_key)", "", ( "These may be one bug rephrased, or rarely distinct bugs sharing a " "lex-first public symbol; filing remains OFF pending JT review " "(JT flag 1)." ), "", ] flagged = [p for p in projections if (p.get("distinct_observation_count") or 0) >= 2] if not flagged: lines.append("No possible collapses.") lines.append("") return "\n".join(lines) for finding in flagged: fk = finding.get("finding_key") or "" lines.extend( [ f"### {str(fk)[:19]}", "", f"- finding_key: {fk}", f"- file: {finding.get('file')}", f"- category: {finding.get('category')}", f"- distinct_observation_count: {finding.get('distinct_observation_count')}", "- collapsed_observations:", ] ) for member in finding.get("collapsed_observations") or []: lines.extend( [ f" - observation_id: {member.get('observation_id')}", f" title: {_truncate_report_value(member.get('title'))}", f" evidence: {_truncate_report_value(member.get('evidence'))}", f" normalized_anchor: {_truncate_report_value(member.get('normalized_anchor'))}", ] ) lines.append("") return "\n".join(lines) def render_report(events: list[dict], ledger: Path, include_diagnostics: bool = False) -> str: """Render a Markdown report. Does NOT print, does NOT write a file. Surfaces CONFIRMED findings (escalations first, then mechanical, then report; newest last_seen within a group), EXCEPT those the human adjudicated as false-positive / ignore-for-now. With include_diagnostics=True, appends a Diagnostics section for findings whose verify_status is in {rejected, abstain, stale, duplicate, unverified} OR that were suppressed by a human adjudication. """ projections = project_findings(events, include_diagnostics=include_diagnostics) # A finding the human adjudicated as false-positive / ignore-for-now is # removed from Confirmed even if the model confirmed it — honoring the # printed append-adjudication commands (otherwise they'd be no-ops). confirmed = [ p for p in projections if p.get("verify_status") == "confirmed" and p.get("human_adjudication") not in _SUPPRESSING_ADJUDICATIONS ] diagnostics = [ p for p in projections if p.get("verify_status") in _DIAGNOSTIC_STATUSES or p.get("human_adjudication") in _SUPPRESSING_ADJUDICATIONS ] parts = ["# Nightwatch Report", ""] parts.append("## Confirmed Findings") parts.append("") if not confirmed: parts.append("No confirmed findings.") parts.append("") else: for finding in confirmed: parts.append(_render_finding_block(finding, ledger)) if include_diagnostics: parts.append("## Diagnostics") parts.append("") if not diagnostics: parts.append("No diagnostic findings.") parts.append("") else: for finding in diagnostics: parts.append(_render_finding_block(finding, ledger)) parts.append(_render_possible_collapse_section(projections)) return "\n".join(parts) def cmd_report(args: argparse.Namespace) -> int: events = load_events(Path(args.ledger)) text = render_report( events, Path(args.ledger), include_diagnostics=args.include_diagnostics ) print(text) return 0 # --------------------------------------------------------------------------- # Argument parsing. # --------------------------------------------------------------------------- def _add_shared_options(parser: argparse.ArgumentParser) -> None: parser.add_argument("--repo-root", default=str(DEFAULT_REPO_ROOT)) parser.add_argument("--ledger", default=str(DEFAULT_LEDGER)) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="nightwatch", description="Phase 0 read-only maintenance CLI (ledger / normalize / ingest).", ) sub = parser.add_subparsers(dest="command", required=True) p_ingest = sub.add_parser("ingest", help="Ingest findings into the append-only ledger.") _add_shared_options(p_ingest) p_ingest.add_argument("--audit-dir", default=str(DEFAULT_AUDIT_DIR)) p_ingest.add_argument("--audit-json", default=None) p_ingest.add_argument("--hygiene-queue", default=str(DEFAULT_HYGIENE_QUEUE)) p_ingest.add_argument("--include-ruff", action="store_true") p_ingest.set_defaults(func=cmd_ingest) p_verify = sub.add_parser("verify", help="Verify unverified observations.") _add_shared_options(p_verify) p_verify.add_argument("--limit", type=int, default=25) p_verify.add_argument("--verify-schema", default=str(DEFAULT_VERIFY_SCHEMA)) p_verify.add_argument("--timeout", type=int, default=3600) p_verify.add_argument("--enqueue-linear", action="store_true") p_verify.set_defaults(func=cmd_verify) p_report = sub.add_parser("report", help="Project the ledger into a Markdown report (read-only).") _add_shared_options(p_report) p_report.add_argument("--include-diagnostics", action="store_true") p_report.set_defaults(func=cmd_report) return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) return args.func(args) if __name__ == "__main__": sys.exit(main())