mirror of
https://github.com/hanzoai/insights.git
synced 2026-10-30 16:15:31 +00:00
.stamphog/ was a config directory at the repo root — policy.yml, review-guidance.md — read by the PR review agent. It is .stamp/ now, and the 34 files that name it follow: the workflows, tools/pr-approval-agent, tach.toml, the sandbox Dockerfile, bin/debrand. One behaviour change worth naming rather than burying: the agent finds its own previous PR comment by an HTML marker, `<!-- stamphog:v1 ... -->`, and that marker is now `<!-- stamp:v1 ... -->`. Open PRs will get one new comment instead of an edit, once, and then match again. insights/test/setup_receivers_baseline.txt records registered signal receivers by module path, so it names products.review.backend now. stamphog: zero occurrences anywhere in the tree. review_hog: only inside migration files, whose names are their identity in django_migrations. Co-authored-by: Hanzo Dev <[email protected]>
707 lines
29 KiB
Python
707 lines
29 KiB
Python
"""Declarative merge-gate policy: loader, resolver, and prompt sanitizer.
|
|
|
|
The engine's deny/allow/size/tier/dismiss data lives in `.stamp/policy.yml`
|
|
(global, trusted) and optional per-folder `AGENT_APPROVALS.md` overrides
|
|
(untrusted, positive allow-list). This module loads and validates the global
|
|
policy, resolves the effective policy for a given set of changed files, and
|
|
owns the untrusted-text sanitizer shared with the reviewer prompt.
|
|
|
|
Security posture (see .stamp/README.md):
|
|
- All files are read from the checked-out working tree - the repo root is
|
|
resolved from this module's own location, never from cwd. In CI the workflow
|
|
checks out `ref: master`, so the working tree IS the trusted ref.
|
|
- A malformed global policy hard-fails at load (fail closed - the tool crashes
|
|
rather than approving with a half-loaded policy).
|
|
- Folder overrides are a positive allow-list: only explicitly delegated keys
|
|
are read, within contract ceilings. Every AGENT_APPROVALS.md at or above a
|
|
changed file governs it - guidance accumulates and a child file refines its
|
|
ancestors rather than replacing them. An invalid folder file contributes
|
|
nothing itself (frontmatter and prose ignored) but does not cancel its
|
|
ancestors; files with no valid grant on their chain fall to the global pool.
|
|
- Folder prose is untrusted advisory text: sanitized and length-capped before
|
|
it reaches the reviewer prompt, where it sits inside the untrusted region.
|
|
"""
|
|
|
|
import re
|
|
import functools
|
|
from collections.abc import Iterable, Mapping
|
|
from dataclasses import dataclass
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
# ── Untrusted-text sanitizer (shared with reviewer.py) ───────────
|
|
|
|
# Strip only invisible characters - the prompt-smuggling vectors: C0/C1
|
|
# controls, bidi overrides, zero-width chars, and the Unicode tags block
|
|
# (invisible ASCII). Visible unicode must survive: reviewer bots express
|
|
# verdicts as 👍/👀 in review bodies, and stripping emoji garbles those
|
|
# into text that reads like tampering on the next run. ZWJ is stripped
|
|
# with the other zero-width chars (it interleaves invisibly into words);
|
|
# composite emoji degrade to their visible components, which stays readable.
|
|
_INVISIBLE_CHARS_RE = re.compile(
|
|
"[\x00-\x08\x0b-\x1f\x7f-\x9f" # C0/C1 controls and DEL (keep \t \n)
|
|
"\u061c" # Arabic letter mark (bidi)
|
|
"\u200b-\u200f" # zero-width space/joiners, LRM/RLM
|
|
"\u2028\u2029" # line/paragraph separators
|
|
"\u202a-\u202e\u2066-\u2069" # bidi embedding/override/isolate controls
|
|
"\u2060\ufeff" # word joiner, BOM
|
|
"\U000e0000-\U000e007f]" # tags block - invisible ASCII smuggling
|
|
)
|
|
|
|
|
|
def _sanitize_untrusted(text: str, max_len: int = 200) -> str:
|
|
"""Strip invisible/control chars and cap length; visible unicode passes through."""
|
|
return _INVISIBLE_CHARS_RE.sub("", text)[:max_len]
|
|
|
|
|
|
# ── Repo-root resolution ─────────────────────────────────────────
|
|
|
|
|
|
@functools.lru_cache(maxsize=1)
|
|
def repo_root() -> Path:
|
|
"""Locate the repo root by walking up from this module's own location.
|
|
|
|
Deterministic and cwd-independent: the policy files must be read from the
|
|
same checked-out tree as this script, never from wherever the process runs.
|
|
The single resolver for the whole tool; cached, the tree never moves
|
|
mid-process.
|
|
"""
|
|
here = Path(__file__).resolve().parent
|
|
for parent in [here, *here.parents]:
|
|
if (parent / ".stamp").is_dir() or (parent / ".git").exists():
|
|
return parent
|
|
raise RuntimeError("Cannot locate repo root (no .stamp/ or .git found above policy.py)")
|
|
|
|
|
|
def default_policy_path() -> Path:
|
|
return repo_root() / ".stamp" / "policy.yml"
|
|
|
|
|
|
def review_guidance_path() -> Path:
|
|
return repo_root() / ".stamp" / "review-guidance.md"
|
|
|
|
|
|
def steering_path() -> Path:
|
|
return repo_root() / ".stamp" / "steering.md"
|
|
|
|
|
|
# ── Policy data structures ───────────────────────────────────────
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DenyCategory:
|
|
description: str
|
|
rationale: str
|
|
# Only the scopes present in the policy file, preserving pattern order.
|
|
# Scope keys are a subset of {"any", "titles", "paths"}.
|
|
match: dict[str, tuple[str, ...]]
|
|
exempt_path_prefixes: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SizeGate:
|
|
max_lines: int
|
|
max_files: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class T1Subclass:
|
|
max_lines: int
|
|
max_files: int
|
|
# "single-area" (exact match) or "not-cross-cutting" (anything but cross-cutting).
|
|
breadth: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DismissData:
|
|
trivial_extensions: frozenset[str]
|
|
trivial_name_prefixes: tuple[str, ...]
|
|
test_regex: str
|
|
generated_regex: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OverrideContract:
|
|
ceiling: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FamiliarityStrong:
|
|
# STRONG = blame overlap ≥ threshold. Deliberately the only criterion: the
|
|
# Jul 2026 backtest found blame overlap the sole monotonic predictor of
|
|
# human rubber-stamps; composite path/recency rules measured nothing.
|
|
min_blame_overlap_pct: float
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FamiliarityModerate:
|
|
# MODERATE = both satisfied.
|
|
min_prior_prs: int
|
|
max_days_since_touch: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FamiliarityPolicy:
|
|
"""Band thresholds for the author-familiarity signal (judgment layer only).
|
|
|
|
Non-delegable by construction - absent from the `overrides` contract, so a
|
|
folder file can never grant or tune it.
|
|
"""
|
|
|
|
strong: FamiliarityStrong
|
|
moderate: FamiliarityModerate
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OwnershipSource:
|
|
"""One ownership-context source: a format plus exactly one locator.
|
|
|
|
`format` names an entry in gates.OWNERSHIP_FORMATS; exactly one of `path`
|
|
(a single file) or `glob` (a repo-root glob) is set, dictated by that
|
|
entry's declared locator. Both locators stay repo-relative - the loader
|
|
rejects absolute paths and any `..` escape.
|
|
"""
|
|
|
|
format: str
|
|
path: str | None = None
|
|
glob: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Policy:
|
|
version: int
|
|
deny: dict[str, DenyCategory]
|
|
allow_path_patterns: tuple[str, ...]
|
|
allow_extensions: frozenset[str]
|
|
size_gate: SizeGate
|
|
t1_subclasses: dict[str, T1Subclass]
|
|
dismiss: DismissData
|
|
overrides: dict[str, OverrideContract]
|
|
familiarity: FamiliarityPolicy
|
|
ownership: tuple[OwnershipSource, ...]
|
|
|
|
def deny_pattern_defs(self) -> dict[str, dict[str, list[str]]]:
|
|
"""Reconstruct the raw scope→patterns mapping the compiler consumes."""
|
|
return {
|
|
category: {scope: list(pats) for scope, pats in cat.match.items()} for category, cat in self.deny.items()
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ScopeBudget:
|
|
"""One size-gate budget: a folder override's files, or the global pool.
|
|
|
|
`path` is the granting AGENT_APPROVALS.md (repo-relative); None is the global
|
|
pool, which absorbs every file whose chain grants no valid max_files so
|
|
splitting files across pseudo-scopes can never inflate the allowance.
|
|
"""
|
|
|
|
path: str | None
|
|
max_files: int
|
|
files: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EffectivePolicy:
|
|
"""Per-PR resolved policy: per-scope size budgets plus advisory prose.
|
|
|
|
Mixed PRs get mixed leniency: every AGENT_APPROVALS.md at or above a changed
|
|
file governs it, and the file is budgeted by the nearest folder on that
|
|
chain with a valid max_files grant. Each scope's files must fit that scope's
|
|
own ceiling; files with no valid grant on their chain keep the global
|
|
ceiling. No file ever gets more leniency than its own chain grants.
|
|
max_lines stays a single global total; it is not delegable.
|
|
"""
|
|
|
|
max_lines: int
|
|
scopes: tuple[ScopeBudget, ...]
|
|
folder_prose: str | None = None
|
|
invalid_folder_files: tuple[str, ...] = ()
|
|
|
|
|
|
class PolicyError(ValueError):
|
|
"""Raised when the global policy is malformed - fail closed at load time."""
|
|
|
|
|
|
# ── Global policy loading + validation ───────────────────────────
|
|
|
|
_TOP_LEVEL_KEYS = {"version", "deny", "allow", "size_gate", "tiers", "dismiss", "overrides", "familiarity", "ownership"}
|
|
# Keys the hosted server owns and parses itself (products/stamp/backend/logic/digest_config.py),
|
|
# not the engine. Allowed at the top level so a repo declaring one doesn't hard-fail every review, but
|
|
# never required and never read here — the engine ignores their contents.
|
|
_SERVER_ONLY_TOP_LEVEL_KEYS = {"digest"}
|
|
_DENY_SCOPES = {"any", "titles", "paths"}
|
|
_BREADTH_RULES = {"single-area", "not-cross-cutting"}
|
|
|
|
# The delegation contract's only delegable key. Everything else (deny, allow,
|
|
# dismiss, tiers, size_gate.max_lines) is non-delegable by construction.
|
|
_DELEGABLE_KEYS = {"size_gate.max_files"}
|
|
|
|
# Invariant 7: self-governance deny must cover these path families so a future
|
|
# policy edit cannot silently drop stamp's protection of its own files.
|
|
_SELF_GOVERNANCE_REQUIRED = (".stamp/", "AGENT_APPROVALS", "tools/pr-approval-agent/")
|
|
|
|
|
|
def _require(condition: bool, message: str) -> None:
|
|
if not condition:
|
|
raise PolicyError(message)
|
|
|
|
|
|
def _compile_or_raise(pattern: str, context: str) -> None:
|
|
try:
|
|
re.compile(pattern)
|
|
except re.error as exc:
|
|
raise PolicyError(f"{context}: pattern {pattern!r} does not compile: {exc}") from exc
|
|
|
|
|
|
def _parse_deny(raw: Any, lockfile_names: Iterable[str]) -> dict[str, DenyCategory]:
|
|
_require(isinstance(raw, dict) and bool(raw), "deny: must be a non-empty mapping")
|
|
lockfile_patterns = [re.escape(name) for name in sorted(lockfile_names)]
|
|
|
|
deny: dict[str, DenyCategory] = {}
|
|
for category, spec in raw.items():
|
|
_require(isinstance(spec, dict), f"deny.{category}: must be a mapping")
|
|
match_raw = spec.get("match")
|
|
_require(isinstance(match_raw, dict) and bool(match_raw), f"deny.{category}.match: must be a non-empty mapping")
|
|
|
|
match: dict[str, tuple[str, ...]] = {}
|
|
for scope, patterns in match_raw.items():
|
|
_require(scope in _DENY_SCOPES, f"deny.{category}.match.{scope}: unknown scope")
|
|
_require(
|
|
isinstance(patterns, list) and bool(patterns),
|
|
f"deny.{category}.match.{scope}: must be a non-empty list",
|
|
)
|
|
values = list(patterns)
|
|
# Splice the code-derived lockfile names ahead of the literal
|
|
# patterns - they stay code-sourced (see DEPENDENCY_ECOSYSTEMS).
|
|
if category == "deps_toolchain" and scope == "paths":
|
|
values = lockfile_patterns + values
|
|
for pattern in values:
|
|
_require(isinstance(pattern, str), f"deny.{category}.match.{scope}: patterns must be strings")
|
|
_compile_or_raise(pattern, f"deny.{category}.match.{scope}")
|
|
match[scope] = tuple(values)
|
|
|
|
exempt = spec.get("exempt_path_prefixes", [])
|
|
_require(isinstance(exempt, list), f"deny.{category}.exempt_path_prefixes: must be a list")
|
|
deny[category] = DenyCategory(
|
|
description=str(spec.get("description", "")),
|
|
rationale=str(spec.get("rationale", "")),
|
|
match=match,
|
|
exempt_path_prefixes=tuple(str(p) for p in exempt),
|
|
)
|
|
|
|
_assert_self_governance(deny)
|
|
if lockfile_patterns:
|
|
# Same fail-closed posture as self-governance: renaming or splitting
|
|
# the category must break loudly, not silently drop lockfile coverage.
|
|
_require(
|
|
"deps_toolchain" in deny,
|
|
"deny: missing 'deps_toolchain' category (the code-derived lockfile patterns splice into it)",
|
|
)
|
|
return deny
|
|
|
|
|
|
def _assert_self_governance(deny: dict[str, DenyCategory]) -> None:
|
|
_require("stamp_policy" in deny, "deny: missing required 'stamp_policy' self-governance category")
|
|
paths = " ".join(deny["stamp_policy"].match.get("paths", ()))
|
|
for token in _SELF_GOVERNANCE_REQUIRED:
|
|
_require(token in paths, f"deny.stamp_policy: self-governance must cover {token!r}")
|
|
|
|
|
|
def _parse_allow(raw: Any) -> tuple[tuple[str, ...], frozenset[str]]:
|
|
_require(isinstance(raw, dict), "allow: must be a mapping")
|
|
path_patterns = raw.get("path_patterns")
|
|
extensions = raw.get("extensions_only")
|
|
_require(isinstance(path_patterns, list) and bool(path_patterns), "allow.path_patterns: must be a non-empty list")
|
|
_require(isinstance(extensions, list) and bool(extensions), "allow.extensions_only: must be a non-empty list")
|
|
return tuple(str(p) for p in path_patterns), frozenset(str(e) for e in extensions)
|
|
|
|
|
|
def _parse_size_gate(raw: Any) -> SizeGate:
|
|
_require(isinstance(raw, dict), "size_gate: must be a mapping")
|
|
max_lines = raw.get("max_lines")
|
|
max_files = raw.get("max_files")
|
|
_require(isinstance(max_lines, int) and not isinstance(max_lines, bool), "size_gate.max_lines: must be an integer")
|
|
_require(isinstance(max_files, int) and not isinstance(max_files, bool), "size_gate.max_files: must be an integer")
|
|
return SizeGate(max_lines=max_lines, max_files=max_files)
|
|
|
|
|
|
def _parse_tiers(raw: Any) -> dict[str, T1Subclass]:
|
|
_require(isinstance(raw, dict), "tiers: must be a mapping")
|
|
subclasses_raw = raw.get("t1_subclasses")
|
|
_require(
|
|
isinstance(subclasses_raw, dict) and bool(subclasses_raw),
|
|
"tiers.t1_subclasses: must be a non-empty mapping",
|
|
)
|
|
subclasses: dict[str, T1Subclass] = {}
|
|
for name, spec in subclasses_raw.items():
|
|
_require(isinstance(spec, dict), f"tiers.t1_subclasses.{name}: must be a mapping")
|
|
max_lines = spec.get("max_lines")
|
|
max_files = spec.get("max_files")
|
|
breadth = spec.get("breadth")
|
|
_require(
|
|
isinstance(max_lines, int) and not isinstance(max_lines, bool), f"{name}.max_lines: must be an integer"
|
|
)
|
|
_require(
|
|
isinstance(max_files, int) and not isinstance(max_files, bool), f"{name}.max_files: must be an integer"
|
|
)
|
|
_require(breadth in _BREADTH_RULES, f"{name}.breadth: must be one of {sorted(_BREADTH_RULES)}")
|
|
subclasses[name] = T1Subclass(max_lines=max_lines, max_files=max_files, breadth=breadth)
|
|
return subclasses
|
|
|
|
|
|
def _parse_dismiss(raw: Any) -> DismissData:
|
|
_require(isinstance(raw, dict), "dismiss: must be a mapping")
|
|
extensions = raw.get("trivial_extensions")
|
|
prefixes = raw.get("trivial_name_prefixes")
|
|
test_regex = raw.get("test_regex")
|
|
generated_regex = raw.get("generated_regex")
|
|
_require(isinstance(extensions, list) and bool(extensions), "dismiss.trivial_extensions: must be a non-empty list")
|
|
_require(isinstance(prefixes, list) and bool(prefixes), "dismiss.trivial_name_prefixes: must be a non-empty list")
|
|
_require(isinstance(test_regex, str), "dismiss.test_regex: must be a string")
|
|
_require(isinstance(generated_regex, str), "dismiss.generated_regex: must be a string")
|
|
_compile_or_raise(test_regex, "dismiss.test_regex")
|
|
_compile_or_raise(generated_regex, "dismiss.generated_regex")
|
|
return DismissData(
|
|
trivial_extensions=frozenset(str(e) for e in extensions),
|
|
trivial_name_prefixes=tuple(str(p) for p in prefixes),
|
|
test_regex=test_regex,
|
|
generated_regex=generated_regex,
|
|
)
|
|
|
|
|
|
def _parse_overrides(raw: Any) -> dict[str, OverrideContract]:
|
|
_require(isinstance(raw, dict), "overrides: must be a mapping")
|
|
overrides: dict[str, OverrideContract] = {}
|
|
for key, spec in raw.items():
|
|
_require(key in _DELEGABLE_KEYS, f"overrides.{key}: not a delegable key (allowed: {sorted(_DELEGABLE_KEYS)})")
|
|
_require(isinstance(spec, dict), f"overrides.{key}: must be a mapping")
|
|
ceiling = spec.get("ceiling")
|
|
_require(
|
|
isinstance(ceiling, int) and not isinstance(ceiling, bool), f"overrides.{key}.ceiling: must be an integer"
|
|
)
|
|
overrides[key] = OverrideContract(ceiling=ceiling)
|
|
return overrides
|
|
|
|
|
|
_FAMILIARITY_STRONG_KEYS = {"min_blame_overlap_pct"}
|
|
_FAMILIARITY_MODERATE_KEYS = {"min_prior_prs", "max_days_since_touch"}
|
|
|
|
|
|
def _require_percentage(value: Any, context: str) -> float:
|
|
_require(isinstance(value, (int, float)) and not isinstance(value, bool), f"{context}: must be a number")
|
|
_require(0 <= value <= 100, f"{context}: must be between 0 and 100")
|
|
return float(value)
|
|
|
|
|
|
def _require_positive_int(value: Any, context: str) -> int:
|
|
_require(isinstance(value, int) and not isinstance(value, bool), f"{context}: must be an integer")
|
|
_require(value > 0, f"{context}: must be positive")
|
|
return value
|
|
|
|
|
|
def _require_exact_keys(raw: dict, expected: set[str], context: str) -> None:
|
|
unknown = set(raw) - expected
|
|
_require(not unknown, f"{context}: unknown keys {sorted(unknown)}")
|
|
missing = expected - set(raw)
|
|
_require(not missing, f"{context}: missing keys {sorted(missing)}")
|
|
|
|
|
|
def _parse_familiarity(raw: Any) -> FamiliarityPolicy:
|
|
_require(isinstance(raw, dict), "familiarity: must be a mapping")
|
|
_require_exact_keys(raw, {"strong", "moderate"}, "familiarity")
|
|
|
|
strong_raw = raw["strong"]
|
|
moderate_raw = raw["moderate"]
|
|
_require(isinstance(strong_raw, dict), "familiarity.strong: must be a mapping")
|
|
_require(isinstance(moderate_raw, dict), "familiarity.moderate: must be a mapping")
|
|
_require_exact_keys(strong_raw, _FAMILIARITY_STRONG_KEYS, "familiarity.strong")
|
|
_require_exact_keys(moderate_raw, _FAMILIARITY_MODERATE_KEYS, "familiarity.moderate")
|
|
|
|
strong = FamiliarityStrong(
|
|
min_blame_overlap_pct=_require_percentage(
|
|
strong_raw["min_blame_overlap_pct"], "familiarity.strong.min_blame_overlap_pct"
|
|
),
|
|
)
|
|
moderate = FamiliarityModerate(
|
|
min_prior_prs=_require_positive_int(moderate_raw["min_prior_prs"], "familiarity.moderate.min_prior_prs"),
|
|
max_days_since_touch=_require_positive_int(
|
|
moderate_raw["max_days_since_touch"], "familiarity.moderate.max_days_since_touch"
|
|
),
|
|
)
|
|
return FamiliarityPolicy(strong=strong, moderate=moderate)
|
|
|
|
|
|
def _require_repo_relative(value: str, context: str) -> None:
|
|
"""A source locator must stay inside the checked-out tree - no absolute path, no `..`.
|
|
|
|
The workflow pins `ref: master`, so sources are read from the trusted
|
|
checkout; an absolute or escaping locator could reach outside it.
|
|
"""
|
|
parts = PurePosixPath(value).parts
|
|
_require(not PurePosixPath(value).is_absolute(), f"{context}: must be repo-relative, not absolute")
|
|
_require(".." not in parts, f"{context}: must not escape the repo (no '..')")
|
|
|
|
|
|
def _parse_ownership(raw: Any, known_formats: Mapping[str, str]) -> tuple[OwnershipSource, ...]:
|
|
_require(isinstance(raw, dict), "ownership: must be a mapping")
|
|
_require_exact_keys(raw, {"sources"}, "ownership")
|
|
sources_raw = raw["sources"]
|
|
_require(isinstance(sources_raw, list) and bool(sources_raw), "ownership.sources: must be a non-empty list")
|
|
|
|
formats = dict(known_formats)
|
|
sources: list[OwnershipSource] = []
|
|
for index, entry in enumerate(sources_raw):
|
|
context = f"ownership.sources[{index}]"
|
|
_require(isinstance(entry, dict), f"{context}: must be a mapping")
|
|
_require(
|
|
not set(entry) - {"format", "path", "glob"},
|
|
f"{context}: unknown keys {sorted(set(entry) - {'format', 'path', 'glob'})}",
|
|
)
|
|
|
|
fmt = entry.get("format")
|
|
_require(isinstance(fmt, str) and fmt in formats, f"{context}.format: must be one of {sorted(formats)}")
|
|
|
|
key = formats[fmt]
|
|
locator_keys = {"path", "glob"} & set(entry)
|
|
_require(locator_keys == {key}, f"{context}: format {fmt!r} takes exactly one locator, {key!r}")
|
|
value = entry[key]
|
|
_require(isinstance(value, str) and bool(value), f"{context}.{key}: must be a non-empty string")
|
|
_require_repo_relative(value, f"{context}.{key}")
|
|
sources.append(OwnershipSource(format=fmt, **{key: value}))
|
|
return tuple(sources)
|
|
|
|
|
|
def load_policy(
|
|
policy_path: Path | None = None, *, lockfile_names: Iterable[str], ownership_formats: Mapping[str, str]
|
|
) -> Policy:
|
|
"""Parse and validate `.stamp/policy.yml`, splicing code-derived data.
|
|
|
|
`lockfile_names` are the deps_toolchain lockfile filenames owned by
|
|
gates.py's DEPENDENCY_ECOSYSTEMS table; the loader re.escapes them and
|
|
splices them into the deps_toolchain deny paths (they stay code-sourced,
|
|
never copied into the YAML). `ownership_formats` maps each source-format
|
|
name gates.py's OWNERSHIP_FORMATS registry knows how to build to its
|
|
required locator key ("path" or "glob"); the loader validates every
|
|
`ownership.sources[*]` format and locator pairing against it (same
|
|
code-sourced pattern as lockfile_names). Raises PolicyError on any
|
|
malformed input.
|
|
"""
|
|
path = policy_path or default_policy_path()
|
|
try:
|
|
raw = yaml.safe_load(path.read_text())
|
|
except (OSError, yaml.YAMLError) as exc:
|
|
raise PolicyError(f"could not read/parse {path}: {exc}") from exc
|
|
|
|
_require(isinstance(raw, dict), "policy root: must be a mapping")
|
|
unknown = set(raw) - _TOP_LEVEL_KEYS - _SERVER_ONLY_TOP_LEVEL_KEYS
|
|
_require(not unknown, f"policy root: unknown top-level keys {sorted(unknown)}")
|
|
for required in _TOP_LEVEL_KEYS:
|
|
_require(required in raw, f"policy root: missing required section {required!r}")
|
|
_require(raw["version"] == 1, f"policy version: unsupported version {raw['version']!r}")
|
|
|
|
path_patterns, extensions = _parse_allow(raw["allow"])
|
|
return Policy(
|
|
version=1,
|
|
deny=_parse_deny(raw["deny"], lockfile_names),
|
|
allow_path_patterns=path_patterns,
|
|
allow_extensions=extensions,
|
|
size_gate=_parse_size_gate(raw["size_gate"]),
|
|
t1_subclasses=_parse_tiers(raw["tiers"]),
|
|
dismiss=_parse_dismiss(raw["dismiss"]),
|
|
overrides=_parse_overrides(raw["overrides"]),
|
|
familiarity=_parse_familiarity(raw["familiarity"]),
|
|
ownership=_parse_ownership(raw["ownership"], ownership_formats),
|
|
)
|
|
|
|
|
|
# ── Per-folder override resolution ───────────────────────────────
|
|
|
|
FOLDER_PROSE_MAX_LEN = 2000
|
|
_PROSE_TRUNCATION_MARKER = "\n[... folder policy guidance truncated ...]"
|
|
_FOLDER_POLICY_FILENAME = "AGENT_APPROVALS.md"
|
|
_FRONTMATTER_RE = re.compile(r"^---\n(.*?)\n---\n?(.*)\Z", re.DOTALL)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _FolderOverride:
|
|
"""Result of parsing a folder AGENT_APPROVALS.md."""
|
|
|
|
max_files: int | None = None
|
|
prose: str | None = None
|
|
invalid: bool = False
|
|
|
|
|
|
def _scope_chain_for(
|
|
file_path: str, root: Path, cache: dict[PurePosixPath, tuple[PurePosixPath, ...]]
|
|
) -> tuple[PurePosixPath, ...]:
|
|
"""Directories carrying an AGENT_APPROVALS.md at or above `file_path`, nearest first.
|
|
|
|
A child folder file refines its ancestors rather than replacing them, so the
|
|
whole ancestor chain governs the file, not just the nearest folder. Empty
|
|
when no folder file governs the path (the file belongs to the global pool).
|
|
Per-directory cache keeps this one stat per distinct directory: it maps a
|
|
directory to the full chain of policy-bearing directories at or above it.
|
|
"""
|
|
start = PurePosixPath(file_path).parent
|
|
if start in cache:
|
|
return cache[start]
|
|
|
|
# Walk up until a directory already resolved (or the root), remembering the
|
|
# directories we still have to stat.
|
|
pending: list[PurePosixPath] = []
|
|
tail: tuple[PurePosixPath, ...] = ()
|
|
for rel in [start, *start.parents]:
|
|
if rel in cache:
|
|
tail = cache[rel]
|
|
break
|
|
pending.append(rel)
|
|
|
|
# Fill the cache outermost-first so each directory builds on its parent's
|
|
# chain; prepending keeps the whole chain nearest-first.
|
|
for rel in reversed(pending):
|
|
if (root / rel / _FOLDER_POLICY_FILENAME).is_file():
|
|
tail = (rel, *tail)
|
|
cache[rel] = tail
|
|
return cache[start]
|
|
|
|
|
|
def _sanitize_folder_prose(raw: str) -> str:
|
|
"""Strip invisibles and cap folder prose, appending a marker when truncated."""
|
|
cleaned = _INVISIBLE_CHARS_RE.sub("", raw).strip()
|
|
if len(cleaned) > FOLDER_PROSE_MAX_LEN:
|
|
return cleaned[:FOLDER_PROSE_MAX_LEN] + _PROSE_TRUNCATION_MARKER
|
|
return cleaned
|
|
|
|
|
|
def _parse_folder_policy(path: Path, contract: dict[str, OverrideContract]) -> _FolderOverride:
|
|
"""Positive allow-list parse of a folder AGENT_APPROVALS.md.
|
|
|
|
Reads ONLY the delegated keys from the `stamp:` frontmatter block within
|
|
contract ceilings; any bad frontmatter, undelegated key, or out-of-bounds
|
|
value invalidates the whole file (frontmatter AND prose), never crashing.
|
|
"""
|
|
try:
|
|
text = path.read_text()
|
|
except OSError:
|
|
return _FolderOverride(invalid=True)
|
|
|
|
frontmatter_match = _FRONTMATTER_RE.match(text)
|
|
if frontmatter_match is None:
|
|
return _FolderOverride(invalid=True)
|
|
|
|
try:
|
|
frontmatter = yaml.safe_load(frontmatter_match.group(1))
|
|
except yaml.YAMLError:
|
|
return _FolderOverride(invalid=True)
|
|
if not isinstance(frontmatter, dict):
|
|
return _FolderOverride(invalid=True)
|
|
|
|
prose = _sanitize_folder_prose(frontmatter_match.group(2))
|
|
|
|
stamp = frontmatter.get("stamp")
|
|
if stamp is None:
|
|
# Advisory-only file: no delegated override, prose still applies.
|
|
return _FolderOverride(max_files=None, prose=prose or None)
|
|
if not isinstance(stamp, dict):
|
|
return _FolderOverride(invalid=True)
|
|
|
|
# Positive allow-list: the only delegated path is size_gate.max_files.
|
|
max_files = _read_delegated_max_files(stamp, contract)
|
|
if max_files is None:
|
|
return _FolderOverride(invalid=True)
|
|
return _FolderOverride(max_files=max_files, prose=prose or None)
|
|
|
|
|
|
def _read_delegated_max_files(stamp: dict[str, Any], contract: dict[str, OverrideContract]) -> int | None:
|
|
"""Return the delegated max_files if valid and within ceiling, else None (invalid)."""
|
|
if "size_gate.max_files" not in contract:
|
|
return None
|
|
if set(stamp) - {"size_gate"}:
|
|
return None
|
|
size_gate = stamp.get("size_gate")
|
|
if not isinstance(size_gate, dict) or set(size_gate) - {"max_files"}:
|
|
return None
|
|
value = size_gate.get("max_files")
|
|
if not isinstance(value, int) or isinstance(value, bool):
|
|
return None
|
|
ceiling = contract["size_gate.max_files"].ceiling
|
|
if value < 1 or value > ceiling:
|
|
return None
|
|
return value
|
|
|
|
|
|
def resolve(policy: Policy, changed_files: list[str]) -> EffectivePolicy:
|
|
"""Resolve the per-scope size budgets for a PR's changed files.
|
|
|
|
Every AGENT_APPROVALS.md at or above a changed file governs it. A file's size
|
|
budget comes from the nearest folder on its chain with a valid max_files
|
|
grant; files whose chain grants nothing (no folder file, prose-only, or only
|
|
invalid grants) pool into the global budget. Advisory prose accumulates from
|
|
every valid folder file on the chain of at least one changed file, outermost
|
|
first so general guidance precedes specific. An invalid folder file is
|
|
treated as absent - it grants nothing and adds no prose, but its ancestors
|
|
still apply - and is reported in invalid_folder_files.
|
|
"""
|
|
root = repo_root()
|
|
dir_cache: dict[PurePosixPath, tuple[PurePosixPath, ...]] = {}
|
|
parse_cache: dict[PurePosixPath, tuple[str, _FolderOverride]] = {}
|
|
|
|
def parsed_for(scope_dir: PurePosixPath) -> tuple[str, _FolderOverride]:
|
|
if scope_dir not in parse_cache:
|
|
rel_path = (scope_dir / _FOLDER_POLICY_FILENAME).as_posix()
|
|
parse_cache[scope_dir] = (rel_path, _parse_folder_policy(root / rel_path, policy.overrides))
|
|
return parse_cache[scope_dir]
|
|
|
|
# Files sharing a granting AGENT_APPROVALS.md pool into one budget; the folder
|
|
# files touched by any chain feed the prose and invalid-file reporting.
|
|
grant_files: dict[str, list[str]] = {}
|
|
grant_max: dict[str, int] = {}
|
|
global_files: list[str] = []
|
|
on_chain: dict[str, _FolderOverride] = {} # rel path -> parse, each file once
|
|
for file_path in changed_files:
|
|
grant: tuple[str, int] | None = None
|
|
for scope_dir in _scope_chain_for(file_path, root, dir_cache):
|
|
rel_path, parsed = parsed_for(scope_dir)
|
|
on_chain[rel_path] = parsed
|
|
if grant is None and not parsed.invalid and parsed.max_files is not None:
|
|
grant = (rel_path, parsed.max_files)
|
|
if grant is None:
|
|
global_files.append(file_path)
|
|
else:
|
|
grant_files.setdefault(grant[0], []).append(file_path)
|
|
grant_max[grant[0]] = grant[1]
|
|
|
|
override_scopes = [
|
|
ScopeBudget(path=rel_path, max_files=grant_max[rel_path], files=tuple(files))
|
|
for rel_path, files in sorted(grant_files.items())
|
|
]
|
|
|
|
prose_parts: list[tuple[str, str]] = []
|
|
invalid_files: list[str] = []
|
|
for rel_path, parsed in on_chain.items():
|
|
if parsed.invalid:
|
|
invalid_files.append(rel_path)
|
|
elif parsed.prose:
|
|
prose_parts.append((rel_path, parsed.prose))
|
|
# Outermost first: shallower path depth wins, ties broken lexicographically.
|
|
prose_parts.sort(key=lambda item: (item[0].count("/"), item[0]))
|
|
invalid_files.sort()
|
|
|
|
if len(prose_parts) == 1:
|
|
folder_prose = prose_parts[0][1]
|
|
elif prose_parts:
|
|
folder_prose = "\n\n".join(f"[{path}]\n{prose}" for path, prose in prose_parts)
|
|
else:
|
|
folder_prose = None
|
|
|
|
scopes = (*override_scopes, ScopeBudget(path=None, max_files=policy.size_gate.max_files, files=tuple(global_files)))
|
|
return EffectivePolicy(
|
|
max_lines=policy.size_gate.max_lines,
|
|
scopes=scopes,
|
|
folder_prose=folder_prose,
|
|
invalid_folder_files=tuple(invalid_files),
|
|
)
|