545 lines
19 KiB
Python
545 lines
19 KiB
Python
"""
|
|
calibrate_helpers.py — deterministic helpers for the `:calibrate` skill
|
|
(lifecycle-aware-doc-hygiene change, task 5.4).
|
|
|
|
Stdlib-only, no model. Three responsibilities, each a small
|
|
single-responsibility class taking injected inputs and returning
|
|
JSON-serializable output:
|
|
|
|
1. `ClusterSampler` — groups unmatched paths by path-shape (directory
|
|
prefix + a filename "shape class" where digit runs -> `#` and hex-looking
|
|
runs -> `~`) and emits capped representative samples per cluster.
|
|
2. `RuleReportBuilder` — given proposed rules (glob + lifetime) and the
|
|
repo's file list, assembles the 5-element rule-report data: glob
|
|
verbatim, matched paths (capped sample + total), near-miss boundary
|
|
paths, lifetime/tier, and a `why` placeholder the caller fills in.
|
|
3. `RuleQualityChecker` — deterministic lint: `class_not_path` (flags globs
|
|
whose filename segment looks instance-unique: long digit runs, hex
|
|
hashes, bare timestamps, or a wildcard-free glob matching exactly one
|
|
existing path) and `prefer_narrower` (compares two candidate globs by
|
|
match-count on the actual tree; the narrower one wins ties).
|
|
|
|
None of these classes call an LLM, read history, or touch git. They operate
|
|
purely on lists of strings (paths) handed in by the orchestrating skill.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Optional
|
|
|
|
import rulebook as _rulebook
|
|
|
|
_DEFAULT_SAMPLE_CAP = 5
|
|
|
|
# --- shape-class helpers ---------------------------------------------------
|
|
|
|
_DIGIT_RUN = re.compile(r"\d+")
|
|
# hex run: 6+ consecutive hex chars (avoids false-triggering on short words)
|
|
_HEX_RUN = re.compile(r"(?<![0-9a-fA-F])[0-9a-fA-F]{6,}(?![0-9a-fA-F])")
|
|
|
|
# "instance-unique" detectors used by both shape-classing and rule-quality
|
|
# lints: a long digit run (8+, e.g. a timestamp or run-id), or a hex run
|
|
# (6+ chars, e.g. a git sha / hash).
|
|
_LONG_DIGIT_RUN = re.compile(r"\d{6,}")
|
|
|
|
|
|
def _filename_shape(name: str) -> str:
|
|
"""Collapse a filename into a shape class: digit runs -> '#', hex runs
|
|
(6+ hex chars) -> '~'. Order matters: hex collapse first (a hex run may
|
|
contain digits that would otherwise be independently collapsed)."""
|
|
shaped = _HEX_RUN.sub("~", name)
|
|
shaped = _DIGIT_RUN.sub("#", shaped)
|
|
return shaped
|
|
|
|
|
|
def _dir_prefix(path: str) -> str:
|
|
parts = PurePosixPath(path).parts[:-1]
|
|
return "/".join(parts)
|
|
|
|
|
|
@dataclass
|
|
class Cluster:
|
|
key: str # "<dir_prefix>::<shape>"
|
|
dir_prefix: str
|
|
shape: str
|
|
paths: list # all paths in this cluster, sorted
|
|
sample: list # capped representative sample
|
|
|
|
@property
|
|
def total(self) -> int:
|
|
return len(self.paths)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"key": self.key,
|
|
"dir_prefix": self.dir_prefix,
|
|
"shape": self.shape,
|
|
"total": self.total,
|
|
"sample": list(self.sample),
|
|
}
|
|
|
|
|
|
class ClusterSampler:
|
|
"""Groups unmatched paths by path-shape (directory prefix + filename
|
|
shape class) and emits per-cluster representative samples, capped."""
|
|
|
|
def __init__(self, sample_cap: int = _DEFAULT_SAMPLE_CAP) -> None:
|
|
self._cap = sample_cap
|
|
|
|
def cluster(self, unmatched_paths: list) -> list:
|
|
"""Returns a list of Cluster, sorted by cluster key for determinism.
|
|
Each cluster groups paths sharing the same directory prefix AND the
|
|
same filename shape class."""
|
|
groups: dict = {}
|
|
for path in unmatched_paths:
|
|
dir_prefix = _dir_prefix(path)
|
|
filename = PurePosixPath(path).name
|
|
shape = _filename_shape(filename)
|
|
key = f"{dir_prefix}::{shape}"
|
|
groups.setdefault(key, []).append(path)
|
|
|
|
clusters = []
|
|
for key in sorted(groups.keys()):
|
|
paths = sorted(groups[key])
|
|
dir_prefix, shape = key.split("::", 1)
|
|
clusters.append(
|
|
Cluster(
|
|
key=key,
|
|
dir_prefix=dir_prefix,
|
|
shape=shape,
|
|
paths=paths,
|
|
sample=paths[: self._cap],
|
|
)
|
|
)
|
|
return clusters
|
|
|
|
def cluster_to_dicts(self, unmatched_paths: list) -> list:
|
|
return [c.to_dict() for c in self.cluster(unmatched_paths)]
|
|
|
|
|
|
# --- rule report assembly ---------------------------------------------------
|
|
|
|
|
|
def _glob_matches(glob_pattern: str, path: str) -> bool:
|
|
return fnmatch.fnmatch(path, glob_pattern)
|
|
|
|
|
|
def _relax_glob(glob_pattern: str) -> Optional[str]:
|
|
"""Produce a strictly broader variant of a glob for near-miss detection.
|
|
|
|
Widens every path segment that already contains a wildcard character to
|
|
a bare '*' (fully generic for that segment), preserving fixed segments
|
|
and the final segment's extension when present. This surfaces exactly
|
|
the #45 boundary-bug class: a glob like `autoresearch/classic-*/` that
|
|
silently misses a sibling `autoresearch/improve-*/` — relaxing the
|
|
`classic-*` segment to `*` reveals the sibling as a near-miss.
|
|
|
|
Returns None if no segment has a wildcard to relax (fully-fixed glob has
|
|
no meaningful near-miss boundary)."""
|
|
parts = glob_pattern.split("/")
|
|
if not parts:
|
|
return None
|
|
|
|
relaxed_parts = list(parts)
|
|
changed = False
|
|
for i, part in enumerate(parts):
|
|
if part in ("*", "**"):
|
|
continue # already fully generic
|
|
if any(ch in part for ch in "*?[") :
|
|
if "." in part:
|
|
stem, _, ext = part.rpartition(".")
|
|
relaxed = "*." + ext
|
|
else:
|
|
relaxed = "*"
|
|
if relaxed != part:
|
|
relaxed_parts[i] = relaxed
|
|
changed = True
|
|
|
|
if not changed:
|
|
return None
|
|
return "/".join(relaxed_parts)
|
|
|
|
|
|
@dataclass
|
|
class RuleReportEntry:
|
|
glob: str
|
|
lifetime: str
|
|
tier: str
|
|
matched_sample: list
|
|
matched_total: int
|
|
near_miss: list
|
|
why: str = ""
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"glob": self.glob,
|
|
"lifetime": self.lifetime,
|
|
"tier": self.tier,
|
|
"matches": {
|
|
"sample": list(self.matched_sample),
|
|
"total": self.matched_total,
|
|
},
|
|
"near_miss": list(self.near_miss),
|
|
"why": self.why,
|
|
}
|
|
|
|
|
|
class RuleReportBuilder:
|
|
"""Given proposed rules and the repo's file list, produces the 5-element
|
|
rule-report data per proposed rule: glob verbatim, matched paths (capped
|
|
sample + total), near-miss boundary, lifetime + tier, and a why string
|
|
(caller-supplied; this class does not author prose)."""
|
|
|
|
def __init__(self, sample_cap: int = _DEFAULT_SAMPLE_CAP) -> None:
|
|
self._cap = sample_cap
|
|
|
|
def build(self, proposed_rule: dict, all_paths: list) -> RuleReportEntry:
|
|
glob_pattern = proposed_rule["glob"]
|
|
lifetime = proposed_rule.get("lifetime", "keep")
|
|
tier = proposed_rule.get("tier", "confirm")
|
|
why = proposed_rule.get("why", "")
|
|
|
|
matched = sorted(p for p in all_paths if _glob_matches(glob_pattern, p))
|
|
|
|
near_miss: list = []
|
|
relaxed = _relax_glob(glob_pattern)
|
|
if relaxed is not None:
|
|
matched_set = set(matched)
|
|
near_miss = sorted(
|
|
p
|
|
for p in all_paths
|
|
if p not in matched_set and _glob_matches(relaxed, p)
|
|
)
|
|
|
|
return RuleReportEntry(
|
|
glob=glob_pattern,
|
|
lifetime=lifetime,
|
|
tier=tier,
|
|
matched_sample=matched[: self._cap],
|
|
matched_total=len(matched),
|
|
near_miss=near_miss,
|
|
why=why,
|
|
)
|
|
|
|
def build_all(self, proposed_rules: list, all_paths: list) -> list:
|
|
return [self.build(r, all_paths).to_dict() for r in proposed_rules]
|
|
|
|
|
|
# --- rule-quality checks -----------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class QualityFinding:
|
|
check: str # "class_not_path" | "prefer_narrower"
|
|
glob: str
|
|
passed: bool
|
|
reason: str
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"check": self.check,
|
|
"glob": self.glob,
|
|
"passed": self.passed,
|
|
"reason": self.reason,
|
|
}
|
|
|
|
|
|
class RuleQualityChecker:
|
|
"""Deterministic lints applied to a proposed glob before persistence:
|
|
|
|
- `class_not_path`: flags a glob whose final path segment looks
|
|
instance-unique (a long digit run / bare timestamp, or a hex-looking
|
|
hash run), OR a glob with zero wildcard characters that currently
|
|
matches exactly one existing path (a rule that can, by construction,
|
|
only ever match that one file — a failed generalization).
|
|
- `prefer_narrower`: given two candidate globs for the same cluster,
|
|
report which one is narrower by match-count against the actual tree
|
|
(ties broken by raw pattern length, longer = more specific = narrower).
|
|
"""
|
|
|
|
def class_not_path(self, glob_pattern: str, all_paths: Optional[list] = None) -> QualityFinding:
|
|
last_segment = glob_pattern.split("/")[-1]
|
|
|
|
if _LONG_DIGIT_RUN.search(last_segment):
|
|
return QualityFinding(
|
|
check="class_not_path",
|
|
glob=glob_pattern,
|
|
passed=False,
|
|
reason=(
|
|
"glob's final segment contains a long digit run "
|
|
"(looks like a run-id or timestamp unique to one instance)"
|
|
),
|
|
)
|
|
|
|
if _HEX_RUN.search(last_segment):
|
|
return QualityFinding(
|
|
check="class_not_path",
|
|
glob=glob_pattern,
|
|
passed=False,
|
|
reason=(
|
|
"glob's final segment contains a hex-looking run "
|
|
"(looks like a hash unique to one instance)"
|
|
),
|
|
)
|
|
|
|
has_wildcard = any(ch in glob_pattern for ch in "*?[")
|
|
if not has_wildcard and all_paths is not None:
|
|
matches = [p for p in all_paths if _glob_matches(glob_pattern, p)]
|
|
if len(matches) <= 1:
|
|
return QualityFinding(
|
|
check="class_not_path",
|
|
glob=glob_pattern,
|
|
passed=False,
|
|
reason=(
|
|
"glob has no wildcard and matches at most one existing "
|
|
"path — by construction it can never match a future "
|
|
"file (failed generalization), flagged loudly"
|
|
),
|
|
)
|
|
|
|
return QualityFinding(
|
|
check="class_not_path",
|
|
glob=glob_pattern,
|
|
passed=True,
|
|
reason="glob names a recurring class, not a single instance",
|
|
)
|
|
|
|
def prefer_narrower(self, glob_a: str, glob_b: str, all_paths: list) -> dict:
|
|
"""Returns {"narrower": <glob>, "match_counts": {a: n, b: n}}."""
|
|
count_a = sum(1 for p in all_paths if _glob_matches(glob_a, p))
|
|
count_b = sum(1 for p in all_paths if _glob_matches(glob_b, p))
|
|
|
|
if count_a != count_b:
|
|
narrower = glob_a if count_a < count_b else glob_b
|
|
else:
|
|
# tie-break on identical match counts: fewer wildcard characters
|
|
# is more specific/narrower; if that also ties, the longer raw
|
|
# pattern is treated as more specific.
|
|
def _wildcard_count(g: str) -> int:
|
|
return sum(g.count(ch) for ch in "*?[")
|
|
|
|
wc_a, wc_b = _wildcard_count(glob_a), _wildcard_count(glob_b)
|
|
if wc_a != wc_b:
|
|
narrower = glob_a if wc_a < wc_b else glob_b
|
|
else:
|
|
narrower = glob_a if len(glob_a) >= len(glob_b) else glob_b
|
|
|
|
return {
|
|
"narrower": narrower,
|
|
"match_counts": {glob_a: count_a, glob_b: count_b},
|
|
}
|
|
|
|
|
|
# --- canonical rules-file writer --------------------------------------------
|
|
|
|
# Known fields for a rule entry — reuses rulebook.py's schema so the writer
|
|
# and the loader never drift (rulebook.py itself stays nomination-unaware;
|
|
# this is a read-only reference to its field set, design.md Non-Goal 1).
|
|
_RULE_KNOWN_FIELDS = _rulebook._KNOWN_FIELDS
|
|
|
|
# Nominations entry schemas (lifecycle-spec.md §2). Consult entries
|
|
# deliberately carry no lifetime; rejected entries deliberately carry no
|
|
# status field.
|
|
_CONSULT_KNOWN_FIELDS = {"glob", "question", "evidence", "cluster_key", "asked_on"}
|
|
_REJECTED_KNOWN_FIELDS = {
|
|
"glob",
|
|
"lifetime",
|
|
"why",
|
|
"consider_instead",
|
|
"rejected_by",
|
|
"judged_on",
|
|
}
|
|
|
|
# Canonical tier order for `rules`: delete-once-served -> temporary -> keep.
|
|
# Anything else (e.g. a missing/unrecognized lifetime) sorts last.
|
|
_LIFETIME_TIER_ORDER = {
|
|
"delete-once-served": 0,
|
|
"temporary": 1,
|
|
"keep": 2,
|
|
}
|
|
|
|
|
|
def _warn_unknown_fields(entry: dict, known_fields: set, label: str, warnings: list) -> None:
|
|
unknown = set(entry.keys()) - known_fields
|
|
if unknown:
|
|
glob_pattern = entry.get("glob", "<no glob>")
|
|
warnings.append(
|
|
f"{label} {glob_pattern!r}: unrecognized field(s) {sorted(unknown)} "
|
|
"— preserved, not dropped"
|
|
)
|
|
|
|
|
|
def _sort_by_glob(entries: list) -> list:
|
|
return sorted(entries, key=lambda e: e.get("glob", ""))
|
|
|
|
|
|
class RulesFileWriter:
|
|
"""The single canonical serialization path for `.dochygiene-rules.json`
|
|
(design.md Implementation shape). Owns:
|
|
|
|
- canonical ordering: `rules` grouped by lifetime tier (delete-once-
|
|
served -> temporary -> keep), glob-sorted within each group;
|
|
`nominations` after `rules`, with `consults` before `rejected`, each
|
|
glob-sorted.
|
|
- idempotency: canonicalizing an already-canonical structure is a no-op.
|
|
- unknown-field discipline: unrecognized fields on rules AND nomination
|
|
entries are warned about and round-tripped, never dropped.
|
|
- the nominations read/validate/warn path — `rulebook.py` stays
|
|
nomination-unaware; this is the only code that interprets the
|
|
`nominations` key.
|
|
|
|
No knowledge of git, classification, or judge/report logic — purely a
|
|
load/canonicalize/write seam over the parsed JSON structure.
|
|
"""
|
|
|
|
def canonicalize(self, data: dict) -> "tuple[dict, list]":
|
|
"""Pure in-memory transform: returns (canonical_dict, warnings).
|
|
Does no I/O."""
|
|
warnings: list = []
|
|
|
|
rules_in = data.get("rules", []) or []
|
|
rules_out = []
|
|
for raw in rules_in:
|
|
rule = dict(raw)
|
|
_warn_unknown_fields(rule, _RULE_KNOWN_FIELDS, "rule", warnings)
|
|
rules_out.append(rule)
|
|
|
|
rules_out = sorted(
|
|
rules_out,
|
|
key=lambda r: (
|
|
_LIFETIME_TIER_ORDER.get(r.get("lifetime"), 99),
|
|
r.get("glob", ""),
|
|
),
|
|
)
|
|
|
|
canonical: dict = {
|
|
"schema_version": data.get("schema_version", 1),
|
|
"rules": rules_out,
|
|
}
|
|
|
|
nominations_in = data.get("nominations") or {}
|
|
consults_in = nominations_in.get("consults", []) or []
|
|
rejected_in = nominations_in.get("rejected", []) or []
|
|
|
|
consults_out = []
|
|
for raw in consults_in:
|
|
entry = dict(raw)
|
|
_warn_unknown_fields(entry, _CONSULT_KNOWN_FIELDS, "consult entry", warnings)
|
|
consults_out.append(entry)
|
|
|
|
rejected_out = []
|
|
for raw in rejected_in:
|
|
entry = dict(raw)
|
|
_warn_unknown_fields(entry, _REJECTED_KNOWN_FIELDS, "rejected entry", warnings)
|
|
rejected_out.append(entry)
|
|
|
|
consults_out = _sort_by_glob(consults_out)
|
|
rejected_out = _sort_by_glob(rejected_out)
|
|
|
|
if consults_out or rejected_out:
|
|
canonical["nominations"] = {
|
|
"consults": consults_out,
|
|
"rejected": rejected_out,
|
|
}
|
|
|
|
return canonical, warnings
|
|
|
|
def load(self, path: Path) -> "tuple[dict, list]":
|
|
"""Reads a rules file from disk and returns (canonical_dict,
|
|
warnings). A missing file yields the empty envelope, matching
|
|
rulebook.py's own missing-file behavior."""
|
|
path = Path(path)
|
|
try:
|
|
text = path.read_text(encoding="utf-8")
|
|
except FileNotFoundError:
|
|
return self.canonicalize({"schema_version": 1, "rules": []})
|
|
|
|
data = json.loads(text)
|
|
return self.canonicalize(data)
|
|
|
|
def write(self, path: Path, data: dict) -> list:
|
|
"""Canonicalizes *data* and writes it to *path* as pretty JSON with
|
|
a trailing newline. Returns the warnings collected during
|
|
canonicalization."""
|
|
path = Path(path)
|
|
canonical, warnings = self.canonicalize(data)
|
|
text = json.dumps(canonical, indent=2) + "\n"
|
|
path.write_text(text, encoding="utf-8")
|
|
return warnings
|
|
|
|
|
|
# --- nomination intake filter ------------------------------------------------
|
|
|
|
|
|
class NominationIntakeFilter:
|
|
"""Deterministic, no-model filter run between haiku nomination and the
|
|
strong-model judge (lifecycle-spec.md §8 step 3; calibrate spec's
|
|
"Deterministic Nomination Intake Filter" requirement).
|
|
|
|
Injected inputs: the rules-file's `rejected` entries and open `consults`
|
|
(both read via `RulesFileWriter`/the calibrate skill, never by this
|
|
class), plus, per call, the haiku nominations and the current scanned
|
|
shortlist. No model, no I/O — pure set/string logic over injected data.
|
|
"""
|
|
|
|
def __init__(self, rejected: list, consults: list) -> None:
|
|
self._rejected = list(rejected)
|
|
self._consults = list(consults)
|
|
|
|
def _matches(self, glob_pattern: str, shortlist: list) -> set:
|
|
return {p for p in shortlist if _glob_matches(glob_pattern, p)}
|
|
|
|
def filter(self, nominations: list, shortlist: list) -> dict:
|
|
"""Returns:
|
|
{
|
|
"survivors": [ {**nomination, "related_rejections": [...]}, ... ],
|
|
"dropped": [ {**nomination, "reason": "..."}, ... ],
|
|
"consults": <all open consults, passed through unconditionally>,
|
|
}
|
|
"""
|
|
survivors = []
|
|
dropped = []
|
|
|
|
for nomination in nominations:
|
|
glob_pattern = nomination.get("glob")
|
|
lifetime = nomination.get("lifetime")
|
|
|
|
exact_match = next(
|
|
(
|
|
r
|
|
for r in self._rejected
|
|
if r.get("glob") == glob_pattern and r.get("lifetime") == lifetime
|
|
),
|
|
None,
|
|
)
|
|
if exact_match is not None:
|
|
dropped.append(
|
|
{
|
|
**nomination,
|
|
"reason": (
|
|
"exact glob+lifetime repeat of a rejected nomination"
|
|
),
|
|
}
|
|
)
|
|
continue
|
|
|
|
related_rejections = []
|
|
nomination_matches = self._matches(glob_pattern, shortlist)
|
|
for rejection in self._rejected:
|
|
rejection_matches = self._matches(rejection.get("glob", ""), shortlist)
|
|
if nomination_matches & rejection_matches:
|
|
related_rejections.append(dict(rejection))
|
|
|
|
survivors.append({**nomination, "related_rejections": related_rejections})
|
|
|
|
return {
|
|
"survivors": survivors,
|
|
"dropped": dropped,
|
|
"consults": list(self._consults),
|
|
}
|