572 lines
23 KiB
Python
572 lines
23 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
report_builder.py — model-free finalize pass for the doc-hygiene `check` skill.
|
||
|
|
|
||
|
|
This is the deterministic step BETWEEN model classification and
|
||
|
|
`StateStore.write_report` (design D1). The model supplies slim per-file
|
||
|
|
classification PROPOSALS (judgment only); this assembler fills the four fields
|
||
|
|
the model must NOT author and emits a full schema-valid machine report plus a
|
||
|
|
deterministic human-report skeleton.
|
||
|
|
|
||
|
|
The four guardrail fields owned here (never model-authored):
|
||
|
|
- exact_edit.expected_sha256 — sha256 of the CURRENT whole file bytes
|
||
|
|
- is_destructive / is_reversible (deterministic ops) — from KIND_TABLE[kind]
|
||
|
|
- safety_tier — derive_safety_tier(...) imported from
|
||
|
|
validate_report.py (single source of truth)
|
||
|
|
- token_estimate.raw_tokens — from the local token estimator (invariant #6)
|
||
|
|
|
||
|
|
Design references:
|
||
|
|
- openspec/changes/add-check/design.md (D1, D3 step 4, Risks)
|
||
|
|
- openspec/specs/report-schema/spec.md (the frozen contract)
|
||
|
|
|
||
|
|
Key decisions (see design + advisor notes):
|
||
|
|
* expected_sha256 is the hash of the WHOLE file bytes, not the anchored span.
|
||
|
|
The anchors are LINE NUMBERS; an edit outside the span would silently shift
|
||
|
|
the anchored lines, so the guard must cover the whole file (matches the
|
||
|
|
spec's "content hash" wording for the mtime guard).
|
||
|
|
* Per-entry exact_edit.generated_at = the instant THIS file's hash was taken
|
||
|
|
(injected clock). It exists iff expected_sha256 exists — i.e. for
|
||
|
|
anchor-bearing kinds only. insert-frontmatter (has_anchor=False) and
|
||
|
|
generative entries carry neither. (insert-frontmatter is the one span case
|
||
|
|
the design did not specify; raw_tokens is counted over the whole file — a
|
||
|
|
documented deviation.)
|
||
|
|
* tool_version is read from .claude-plugin/plugin.json (never hardcoded).
|
||
|
|
|
||
|
|
Usage (CLI):
|
||
|
|
python report_builder.py --scan <scan.json> --proposals <proposals.json> \\
|
||
|
|
[--out-json <path>] [--out-md <path>]
|
||
|
|
# paths default to stdin (a single JSON object {"scan":..,"proposals":..})
|
||
|
|
# and stdout (a JSON object {"machine_report":.., "human_report":..}).
|
||
|
|
|
||
|
|
Exit codes:
|
||
|
|
0 — built successfully
|
||
|
|
1 — malformed proposal (structured error on stderr)
|
||
|
|
2 — usage / IO error
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Callable, Optional, Protocol
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Import the single source of truth for safety-tier + kind table.
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
_SCRIPTS_DIR = Path(__file__).resolve().parent
|
||
|
|
if str(_SCRIPTS_DIR) not in sys.path:
|
||
|
|
sys.path.insert(0, str(_SCRIPTS_DIR))
|
||
|
|
|
||
|
|
from validate_report import KIND_TABLE, derive_safety_tier # noqa: E402
|
||
|
|
from token_estimator import default_estimator, TokenEstimator # noqa: E402
|
||
|
|
|
||
|
|
|
||
|
|
SCHEMA_VERSION = "1.0"
|
||
|
|
|
||
|
|
_PLUGIN_JSON = _SCRIPTS_DIR.parent / ".claude-plugin" / "plugin.json"
|
||
|
|
|
||
|
|
_VALID_OP_TYPES = ("deterministic", "generative")
|
||
|
|
_STALE_SUBTYPES = {"contradicted", "orphaned", "superseded", "provisional", "completed-in-place", "duplicated"}
|
||
|
|
_BLOAT_SUBTYPES = {"distill", "split", "freeze"}
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Errors
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
class MalformedProposalError(ValueError):
|
||
|
|
"""A classification proposal is structurally invalid (rejected pre-compute)."""
|
||
|
|
|
||
|
|
def __init__(self, index: int, field: str, message: str) -> None:
|
||
|
|
self.index = index
|
||
|
|
self.field = field
|
||
|
|
self.message = message
|
||
|
|
super().__init__(f"proposal[{index}].{field}: {message}")
|
||
|
|
|
||
|
|
def as_dict(self) -> dict:
|
||
|
|
return {"index": self.index, "field": self.field, "message": self.message}
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Injectable seams: clock + file reader
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
class _Clock(Protocol):
|
||
|
|
def now(self) -> datetime: ...
|
||
|
|
|
||
|
|
|
||
|
|
class RealClock:
|
||
|
|
"""Production clock — UTC now()."""
|
||
|
|
|
||
|
|
def now(self) -> datetime:
|
||
|
|
return datetime.now(timezone.utc)
|
||
|
|
|
||
|
|
|
||
|
|
class _FileReader(Protocol):
|
||
|
|
def read_bytes(self, path: Path) -> bytes: ...
|
||
|
|
def read_text(self, path: Path) -> str: ...
|
||
|
|
|
||
|
|
|
||
|
|
class RealFileReader:
|
||
|
|
"""Production reader — raw bytes for hashing, decoded text for spans."""
|
||
|
|
|
||
|
|
def read_bytes(self, path: Path) -> bytes:
|
||
|
|
return Path(path).read_bytes()
|
||
|
|
|
||
|
|
def read_text(self, path: Path) -> str:
|
||
|
|
return Path(path).read_text(encoding="utf-8", errors="replace")
|
||
|
|
|
||
|
|
|
||
|
|
def _iso(dt: datetime) -> str:
|
||
|
|
"""Serialise a datetime as ISO-8601 UTC."""
|
||
|
|
if dt.tzinfo is None:
|
||
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
||
|
|
return dt.isoformat()
|
||
|
|
|
||
|
|
|
||
|
|
def _read_tool_version() -> tuple[str, bool]:
|
||
|
|
"""Return (tool_version, fell_back). Reads .claude-plugin/plugin.json."""
|
||
|
|
try:
|
||
|
|
data = json.loads(_PLUGIN_JSON.read_text(encoding="utf-8"))
|
||
|
|
version = data.get("version")
|
||
|
|
if isinstance(version, str) and version:
|
||
|
|
return version, False
|
||
|
|
except (OSError, json.JSONDecodeError):
|
||
|
|
pass
|
||
|
|
return "0.0.0", True
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Report builder
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
class ReportBuilder:
|
||
|
|
"""Assembles a schema-valid machine report from a scan artifact + proposals.
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
project_root:
|
||
|
|
Resolved project root; entry paths are relative to it and files are
|
||
|
|
read from here.
|
||
|
|
clock:
|
||
|
|
Injected clock. Each file-hash instant and the envelope `generated_at`
|
||
|
|
come from successive `.now()` calls — so per-entry `generated_at`
|
||
|
|
differs from the envelope stamp (use a monotonic clock to observe it).
|
||
|
|
reader:
|
||
|
|
Injected file reader (bytes for hashing, text for span extraction).
|
||
|
|
estimator:
|
||
|
|
Token estimator producing `token_estimate`. Defaults to the local
|
||
|
|
deterministic estimator (no model, no network — invariant #6).
|
||
|
|
tool_version:
|
||
|
|
Override for the plugin version; defaults to reading plugin.json.
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
project_root: Path,
|
||
|
|
clock: Optional[_Clock] = None,
|
||
|
|
reader: Optional[_FileReader] = None,
|
||
|
|
estimator: Optional[TokenEstimator] = None,
|
||
|
|
tool_version: Optional[str] = None,
|
||
|
|
) -> None:
|
||
|
|
self._root = Path(project_root)
|
||
|
|
self._clock = clock or RealClock()
|
||
|
|
self._reader = reader or RealFileReader()
|
||
|
|
self._estimator = estimator or default_estimator()
|
||
|
|
if tool_version is not None:
|
||
|
|
self._tool_version = tool_version
|
||
|
|
self._tool_version_fell_back = False
|
||
|
|
else:
|
||
|
|
self._tool_version, self._tool_version_fell_back = _read_tool_version()
|
||
|
|
|
||
|
|
@property
|
||
|
|
def tool_version_fell_back(self) -> bool:
|
||
|
|
"""True if plugin.json could not be read and a fallback version is used."""
|
||
|
|
return self._tool_version_fell_back
|
||
|
|
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
# Public
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
|
||
|
|
def build(self, scan: dict, proposals: list) -> dict:
|
||
|
|
"""Return {"machine_report": <dict>, "human_report": <str>}.
|
||
|
|
|
||
|
|
Raises MalformedProposalError if any proposal is structurally invalid
|
||
|
|
(validated BEFORE any hashing/IO).
|
||
|
|
"""
|
||
|
|
if not isinstance(scan, dict):
|
||
|
|
raise MalformedProposalError(-1, "(scan)", "scan artifact must be a JSON object")
|
||
|
|
if not isinstance(proposals, list):
|
||
|
|
raise MalformedProposalError(-1, "(proposals)", "proposals must be a JSON array")
|
||
|
|
|
||
|
|
# 1) Validate ALL proposals up-front (reject before computing anything).
|
||
|
|
for i, proposal in enumerate(proposals):
|
||
|
|
self._validate_proposal(proposal, i)
|
||
|
|
|
||
|
|
# 2) Assemble entries (now safe to read files / hash / estimate).
|
||
|
|
entries = [self._build_entry(p) for p in proposals]
|
||
|
|
|
||
|
|
# 3) Envelope. The envelope generated_at is its own clock instant
|
||
|
|
# (the run instant the skill also stamps as last_check).
|
||
|
|
envelope_generated_at = _iso(self._clock.now())
|
||
|
|
scan_block = self._build_scan_block(scan)
|
||
|
|
shortlist = list(scan.get("shortlist", [])) # verbatim
|
||
|
|
|
||
|
|
machine_report = {
|
||
|
|
"schema_version": SCHEMA_VERSION,
|
||
|
|
"tool_version": self._tool_version,
|
||
|
|
"generated_at": envelope_generated_at,
|
||
|
|
"scan": scan_block,
|
||
|
|
"shortlist": shortlist,
|
||
|
|
"entries": entries,
|
||
|
|
}
|
||
|
|
|
||
|
|
human_report = self._build_human_report(
|
||
|
|
machine_report, scan, proposals, envelope_generated_at
|
||
|
|
)
|
||
|
|
|
||
|
|
return {"machine_report": machine_report, "human_report": human_report}
|
||
|
|
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
# Proposal validation (pre-compute)
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
|
||
|
|
def _validate_proposal(self, proposal: Any, idx: int) -> None:
|
||
|
|
if not isinstance(proposal, dict):
|
||
|
|
raise MalformedProposalError(idx, "(root)", "proposal must be a JSON object")
|
||
|
|
|
||
|
|
for key in ("path", "category", "op", "op_type"):
|
||
|
|
if key not in proposal:
|
||
|
|
raise MalformedProposalError(idx, key, f"required field '{key}' is missing")
|
||
|
|
|
||
|
|
path = proposal.get("path")
|
||
|
|
if not isinstance(path, str) or not path:
|
||
|
|
raise MalformedProposalError(idx, "path", "path must be a non-empty string")
|
||
|
|
|
||
|
|
self._validate_category(proposal.get("category"), idx)
|
||
|
|
|
||
|
|
op_type = proposal.get("op_type")
|
||
|
|
if op_type not in _VALID_OP_TYPES:
|
||
|
|
raise MalformedProposalError(
|
||
|
|
idx, "op_type", f"op_type must be one of {list(_VALID_OP_TYPES)}, got {op_type!r}"
|
||
|
|
)
|
||
|
|
|
||
|
|
has_exact_edit = "exact_edit" in proposal and proposal["exact_edit"] is not None
|
||
|
|
|
||
|
|
if op_type == "generative":
|
||
|
|
if has_exact_edit:
|
||
|
|
raise MalformedProposalError(
|
||
|
|
idx, "exact_edit", "generative proposal must NOT carry exact_edit"
|
||
|
|
)
|
||
|
|
# reducible_range is required so we can count raw_tokens on real text.
|
||
|
|
self._validate_range(
|
||
|
|
proposal.get("reducible_range"), idx, "reducible_range", required=True
|
||
|
|
)
|
||
|
|
else: # deterministic
|
||
|
|
if not has_exact_edit:
|
||
|
|
raise MalformedProposalError(
|
||
|
|
idx, "exact_edit", "deterministic proposal MUST carry an exact_edit skeleton"
|
||
|
|
)
|
||
|
|
self._validate_exact_edit_skeleton(proposal["exact_edit"], idx)
|
||
|
|
|
||
|
|
def _validate_category(self, category: Any, idx: int) -> None:
|
||
|
|
if not isinstance(category, dict):
|
||
|
|
raise MalformedProposalError(idx, "category", "category must be an object with class+subtype")
|
||
|
|
cls = category.get("class")
|
||
|
|
subtype = category.get("subtype")
|
||
|
|
if cls not in ("stale", "bloat"):
|
||
|
|
raise MalformedProposalError(idx, "category.class", f"must be 'stale' or 'bloat', got {cls!r}")
|
||
|
|
if cls == "stale" and subtype not in _STALE_SUBTYPES:
|
||
|
|
raise MalformedProposalError(
|
||
|
|
idx, "category.subtype", f"for class 'stale' must be one of {sorted(_STALE_SUBTYPES)}, got {subtype!r}"
|
||
|
|
)
|
||
|
|
if cls == "bloat" and subtype not in _BLOAT_SUBTYPES:
|
||
|
|
raise MalformedProposalError(
|
||
|
|
idx, "category.subtype", f"for class 'bloat' must be one of {sorted(_BLOAT_SUBTYPES)}, got {subtype!r}"
|
||
|
|
)
|
||
|
|
|
||
|
|
def _validate_exact_edit_skeleton(self, exact_edit: Any, idx: int) -> None:
|
||
|
|
if not isinstance(exact_edit, dict):
|
||
|
|
raise MalformedProposalError(idx, "exact_edit", "exact_edit must be a JSON object")
|
||
|
|
kind = exact_edit.get("kind")
|
||
|
|
if kind not in KIND_TABLE:
|
||
|
|
raise MalformedProposalError(
|
||
|
|
idx, "exact_edit.kind", f"unknown kind {kind!r}; valid: {sorted(KIND_TABLE)}"
|
||
|
|
)
|
||
|
|
spec = KIND_TABLE[kind]
|
||
|
|
# Reuse the validator's kind table so input + output rules agree.
|
||
|
|
for req_field in spec.required_fields:
|
||
|
|
if req_field not in exact_edit:
|
||
|
|
raise MalformedProposalError(
|
||
|
|
idx, f"exact_edit.{req_field}", f"kind '{kind}' requires field '{req_field}'"
|
||
|
|
)
|
||
|
|
if spec.has_anchor:
|
||
|
|
self._validate_range(exact_edit.get("anchor"), idx, "exact_edit.anchor", required=True)
|
||
|
|
|
||
|
|
def _validate_range(self, rng: Any, idx: int, field_name: str, required: bool) -> None:
|
||
|
|
if rng is None:
|
||
|
|
if required:
|
||
|
|
raise MalformedProposalError(idx, field_name, f"required range '{field_name}' is missing")
|
||
|
|
return
|
||
|
|
if not isinstance(rng, dict):
|
||
|
|
raise MalformedProposalError(idx, field_name, "range must be an object with start_line+end_line")
|
||
|
|
for k in ("start_line", "end_line"):
|
||
|
|
v = rng.get(k)
|
||
|
|
if not isinstance(v, int) or isinstance(v, bool):
|
||
|
|
raise MalformedProposalError(idx, f"{field_name}.{k}", f"{k} must be an integer")
|
||
|
|
if rng["start_line"] > rng["end_line"]:
|
||
|
|
raise MalformedProposalError(idx, field_name, "start_line must be <= end_line")
|
||
|
|
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
# Entry assembly
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
|
||
|
|
def _build_entry(self, proposal: dict) -> dict:
|
||
|
|
path = proposal["path"]
|
||
|
|
op_type = proposal["op_type"]
|
||
|
|
abs_path = self._root / path
|
||
|
|
|
||
|
|
entry: dict[str, Any] = {
|
||
|
|
"path": path,
|
||
|
|
"category": dict(proposal["category"]),
|
||
|
|
"signals": list(proposal.get("signals", [])),
|
||
|
|
"op": proposal["op"],
|
||
|
|
"op_type": op_type,
|
||
|
|
}
|
||
|
|
|
||
|
|
if op_type == "deterministic":
|
||
|
|
kind = proposal["exact_edit"]["kind"]
|
||
|
|
spec = KIND_TABLE[kind]
|
||
|
|
# Guardrail fields: ALWAYS from the kind table, never from the proposal.
|
||
|
|
entry["is_destructive"] = spec.is_destructive
|
||
|
|
entry["is_reversible"] = spec.is_reversible
|
||
|
|
entry["safety_tier"] = derive_safety_tier(
|
||
|
|
op_type, spec.is_destructive, spec.is_reversible
|
||
|
|
)
|
||
|
|
exact_edit = self._build_exact_edit(proposal["exact_edit"], spec, abs_path)
|
||
|
|
entry["exact_edit"] = exact_edit
|
||
|
|
span_text = self._extract_span(proposal["exact_edit"], kind, abs_path)
|
||
|
|
else: # generative
|
||
|
|
# No exact_edit; (is_destructive, is_reversible) characterize the op.
|
||
|
|
# A generative prose transform preserves history → non-destructive,
|
||
|
|
# reversible; the tier is forced to confirm by op_type regardless.
|
||
|
|
entry["is_destructive"] = False
|
||
|
|
entry["is_reversible"] = True
|
||
|
|
entry["safety_tier"] = derive_safety_tier(op_type, False, True)
|
||
|
|
span_text = self._read_range(abs_path, proposal["reducible_range"])
|
||
|
|
|
||
|
|
entry["token_estimate"] = self._estimator.estimate_for_report(span_text)
|
||
|
|
return entry
|
||
|
|
|
||
|
|
def _build_exact_edit(self, skeleton: dict, spec, abs_path: Path) -> dict:
|
||
|
|
"""Copy the skeleton and stamp the deterministic fields."""
|
||
|
|
kind = skeleton["kind"]
|
||
|
|
exact_edit: dict[str, Any] = {"kind": kind}
|
||
|
|
# Carry the model-supplied kind-specific fields verbatim.
|
||
|
|
for key, value in skeleton.items():
|
||
|
|
if key == "kind":
|
||
|
|
continue
|
||
|
|
exact_edit[key] = value
|
||
|
|
|
||
|
|
if spec.has_anchor:
|
||
|
|
# expected_sha256 = hash of the CURRENT whole file bytes; generated_at
|
||
|
|
# = that hash instant. Both exist iff the kind is anchor-bearing.
|
||
|
|
hash_instant = self._clock.now()
|
||
|
|
file_bytes = self._reader.read_bytes(abs_path)
|
||
|
|
exact_edit["expected_sha256"] = hashlib.sha256(file_bytes).hexdigest()
|
||
|
|
exact_edit["generated_at"] = _iso(hash_instant)
|
||
|
|
return exact_edit
|
||
|
|
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
# Span extraction (for raw_tokens)
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
|
||
|
|
def _extract_span(self, skeleton: dict, kind: str, abs_path: Path) -> str:
|
||
|
|
"""Return the text whose tokens are counted for this deterministic op."""
|
||
|
|
if kind == "move-to-archive":
|
||
|
|
# Whole file is relocated → count the whole file.
|
||
|
|
return self._reader.read_text(abs_path)
|
||
|
|
if kind == "insert-frontmatter":
|
||
|
|
# No anchor; the doc being frozen is the relevant span (whole file).
|
||
|
|
# NOTE: a documented deviation — the design did not specify a span
|
||
|
|
# for insert-frontmatter.
|
||
|
|
return self._reader.read_text(abs_path)
|
||
|
|
# delete-range / replace-text / dedupe — count the anchored span.
|
||
|
|
return self._read_range(abs_path, skeleton["anchor"])
|
||
|
|
|
||
|
|
def _read_range(self, abs_path: Path, rng: dict) -> str:
|
||
|
|
"""Return the inclusive 1-based line range [start_line, end_line]."""
|
||
|
|
text = self._reader.read_text(abs_path)
|
||
|
|
lines = text.splitlines()
|
||
|
|
start = max(1, int(rng["start_line"]))
|
||
|
|
end = min(len(lines), int(rng["end_line"]))
|
||
|
|
if start > end:
|
||
|
|
return ""
|
||
|
|
return "\n".join(lines[start - 1:end])
|
||
|
|
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
# Scan block
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
|
||
|
|
def _build_scan_block(self, scan: dict) -> dict:
|
||
|
|
"""Project the scan artifact onto the schema's `scan` object."""
|
||
|
|
return {
|
||
|
|
"project_root": scan.get("project_root", str(self._root)),
|
||
|
|
"scope_globs": scan.get("scope_globs", []),
|
||
|
|
"excluded_dirs": scan.get("excluded_dirs", []),
|
||
|
|
"files_scanned": scan.get("files_scanned", 0),
|
||
|
|
}
|
||
|
|
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
# Human report skeleton
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
|
||
|
|
def _build_human_report(
|
||
|
|
self,
|
||
|
|
machine_report: dict,
|
||
|
|
scan: dict,
|
||
|
|
proposals: list,
|
||
|
|
generated_at: str,
|
||
|
|
) -> str:
|
||
|
|
entries = machine_report["entries"]
|
||
|
|
shortlist = machine_report["shortlist"]
|
||
|
|
# gloss: optional per-path "why" line a caller may attach to a proposal.
|
||
|
|
gloss_by_path = {
|
||
|
|
p["path"]: p["gloss"]
|
||
|
|
for p in proposals
|
||
|
|
if isinstance(p, dict) and isinstance(p.get("gloss"), str)
|
||
|
|
}
|
||
|
|
|
||
|
|
candidate_paths = {e["path"] for e in entries}
|
||
|
|
cleared = [p for p in shortlist if p not in candidate_paths]
|
||
|
|
|
||
|
|
scope_globs = scan.get("scope_globs", [])
|
||
|
|
lines: list[str] = []
|
||
|
|
lines.append("# doc-hygiene check report")
|
||
|
|
lines.append("")
|
||
|
|
lines.append(f"- generated_at: {generated_at}")
|
||
|
|
lines.append(f"- scope: {', '.join(scope_globs) if scope_globs else '(default)'}")
|
||
|
|
lines.append(f"- files scanned: {scan.get('files_scanned', 0)}")
|
||
|
|
lines.append(f"- candidates: {len(entries)}")
|
||
|
|
lines.append(f"- cleared: {len(cleared)}")
|
||
|
|
lines.append("")
|
||
|
|
|
||
|
|
stale = [e for e in entries if e["category"].get("class") == "stale"]
|
||
|
|
bloat = [e for e in entries if e["category"].get("class") == "bloat"]
|
||
|
|
|
||
|
|
lines.extend(self._render_group("Stale", stale, gloss_by_path))
|
||
|
|
lines.extend(self._render_group("Bloat", bloat, gloss_by_path))
|
||
|
|
|
||
|
|
lines.append("## Cleared")
|
||
|
|
lines.append("")
|
||
|
|
if cleared:
|
||
|
|
for path in cleared:
|
||
|
|
lines.append(f"- {path}")
|
||
|
|
else:
|
||
|
|
lines.append("- (none)")
|
||
|
|
lines.append("")
|
||
|
|
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
def _render_group(self, title: str, group: list, gloss_by_path: dict) -> list:
|
||
|
|
lines = [f"## {title}", ""]
|
||
|
|
if not group:
|
||
|
|
lines.append("- (none)")
|
||
|
|
lines.append("")
|
||
|
|
return lines
|
||
|
|
for e in group:
|
||
|
|
cat = e["category"]
|
||
|
|
subtype = cat.get("subtype", "?")
|
||
|
|
raw = e["token_estimate"].get("raw_tokens", 0)
|
||
|
|
signal_names = ", ".join(
|
||
|
|
s.get("name", "?") for s in e.get("signals", []) if isinstance(s, dict)
|
||
|
|
) or "(none)"
|
||
|
|
lines.append(
|
||
|
|
f"- {e['path']} · {cat.get('class')}/{subtype} · {e['op']} "
|
||
|
|
f"· {e['safety_tier']} · ~{raw} tokens · {signal_names}"
|
||
|
|
)
|
||
|
|
gloss = gloss_by_path.get(e["path"])
|
||
|
|
if gloss:
|
||
|
|
lines.append(f" - {gloss}")
|
||
|
|
lines.append("")
|
||
|
|
return lines
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# CLI
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
def _load_json(path: Optional[str]) -> Any:
|
||
|
|
if path is None or path == "-":
|
||
|
|
return json.load(sys.stdin)
|
||
|
|
with open(path, "r", encoding="utf-8") as fh:
|
||
|
|
return json.load(fh)
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv: Optional[list] = None) -> int:
|
||
|
|
parser = argparse.ArgumentParser(
|
||
|
|
description="doc-hygiene report builder — model-free finalize pass."
|
||
|
|
)
|
||
|
|
parser.add_argument("--scan", default=None, help="Path to scan-artifact JSON ('-' or omit = stdin).")
|
||
|
|
parser.add_argument("--proposals", default=None, help="Path to proposals JSON.")
|
||
|
|
parser.add_argument("--root", default=None, help="Project root (default: scan.project_root).")
|
||
|
|
parser.add_argument("--out-json", default=None, help="Write machine report JSON here (default: stdout bundle).")
|
||
|
|
parser.add_argument("--out-md", default=None, help="Write human report markdown here.")
|
||
|
|
args = parser.parse_args(argv)
|
||
|
|
|
||
|
|
# Input resolution: if neither --scan nor --proposals given, read a single
|
||
|
|
# bundle {"scan":..,"proposals":..} from stdin.
|
||
|
|
try:
|
||
|
|
if args.scan is None and args.proposals is None:
|
||
|
|
bundle = _load_json(None)
|
||
|
|
if not isinstance(bundle, dict) or "scan" not in bundle or "proposals" not in bundle:
|
||
|
|
print(json.dumps({"error": "stdin bundle must have 'scan' and 'proposals'"}), file=sys.stderr)
|
||
|
|
return 2
|
||
|
|
scan = bundle["scan"]
|
||
|
|
proposals = bundle["proposals"]
|
||
|
|
else:
|
||
|
|
scan = _load_json(args.scan)
|
||
|
|
proposals = _load_json(args.proposals)
|
||
|
|
except (OSError, json.JSONDecodeError) as exc:
|
||
|
|
print(json.dumps({"error": f"failed to read input: {exc}"}), file=sys.stderr)
|
||
|
|
return 2
|
||
|
|
|
||
|
|
root = Path(args.root) if args.root else Path(scan.get("project_root", "."))
|
||
|
|
|
||
|
|
builder = ReportBuilder(project_root=root)
|
||
|
|
try:
|
||
|
|
result = builder.build(scan, proposals)
|
||
|
|
except MalformedProposalError as exc:
|
||
|
|
print(json.dumps({"error": "malformed proposal", "detail": exc.as_dict()}), file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
|
||
|
|
machine_report = result["machine_report"]
|
||
|
|
human_report = result["human_report"]
|
||
|
|
|
||
|
|
if builder.tool_version_fell_back:
|
||
|
|
print(
|
||
|
|
json.dumps({"warning": "tool_version fell back; plugin.json unreadable"}),
|
||
|
|
file=sys.stderr,
|
||
|
|
)
|
||
|
|
|
||
|
|
wrote_file = False
|
||
|
|
if args.out_json:
|
||
|
|
Path(args.out_json).write_text(json.dumps(machine_report, indent=2), encoding="utf-8")
|
||
|
|
wrote_file = True
|
||
|
|
if args.out_md:
|
||
|
|
Path(args.out_md).write_text(human_report, encoding="utf-8")
|
||
|
|
wrote_file = True
|
||
|
|
|
||
|
|
if not wrote_file:
|
||
|
|
print(json.dumps({"machine_report": machine_report, "human_report": human_report}, indent=2))
|
||
|
|
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
sys.exit(main())
|