generated from coulomb/repo-seed
tegwick
ab22d22bfb
gating.py: two-tier evidence bar (OQ5) — promote floor (frequency/sessions/ cost_impact) plus a stricter distribution-eligibility floor that sets a promoted pattern to approved+distribution_ready vs provisional. Wired into review() so thin approvals land provisional. bloat_warnings flags duplicate and near-duplicate (same signal-type+locus) candidates (OQ6). [curate]/ [curate.gate] knobs in config.toml. 6 new tests; suite 64/64 green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
118 lines
4.3 KiB
Python
118 lines
4.3 KiB
Python
"""Promotion evidence-bar + bloat guard (design OQ5/OQ6; T04).
|
|
|
|
Two gates protect the catalog:
|
|
|
|
* **Evidence bar (OQ5)** — a candidate must clear configurable floors
|
|
(frequency, distinct supporting sessions) before it may be promoted at all.
|
|
A separate, stricter bar decides whether the promoted pattern is
|
|
*distribution-eligible* (``status="approved"``, ``distribution_ready=True``)
|
|
vs. merely ``provisional`` — the minimum trustworthy evidence before a pattern
|
|
is allowed near live agent environments.
|
|
|
|
* **Bloat guard (OQ6)** — flags candidates that would add little: a duplicate of
|
|
an already-cataloged pattern, or a near-duplicate sharing the same
|
|
signal-type+locus. Keeps the catalog lean so agent context budgets aren't
|
|
degraded by low-value instructions.
|
|
|
|
Knobs live under ``[curate]`` in ``config.toml``; :func:`gate_config` reads them
|
|
with safe defaults so the module also works config-free (tests).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional
|
|
|
|
from .schema import SolutionPattern
|
|
|
|
|
|
@dataclass
|
|
class GateConfig:
|
|
# promotion floor (OQ5)
|
|
min_frequency: int = 2
|
|
min_sessions: int = 2
|
|
min_cost_impact: float = 0.0
|
|
# distribution-eligibility floor (stricter; OQ5)
|
|
dist_require_cross_flavor: bool = False
|
|
dist_min_frequency: int = 3
|
|
dist_min_cost_impact: float = 0.0
|
|
|
|
|
|
def gate_config(config: Optional[dict] = None) -> GateConfig:
|
|
c = (config or {}).get("curate", {}) if config else {}
|
|
g = c.get("gate", {}) if isinstance(c, dict) else {}
|
|
return GateConfig(
|
|
min_frequency=g.get("min_frequency", 2),
|
|
min_sessions=g.get("min_sessions", 2),
|
|
min_cost_impact=g.get("min_cost_impact", 0.0),
|
|
dist_require_cross_flavor=g.get("dist_require_cross_flavor", False),
|
|
dist_min_frequency=g.get("dist_min_frequency", 3),
|
|
dist_min_cost_impact=g.get("dist_min_cost_impact", 0.0),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class GateResult:
|
|
promotable: bool
|
|
distribution_ready: bool
|
|
status: str # "approved" if distribution-ready else "provisional"
|
|
reasons: list = field(default_factory=list)
|
|
|
|
|
|
def _n_sessions(candidate: dict) -> int:
|
|
return len(candidate.get("sessions", []) or [])
|
|
|
|
|
|
def evaluate(candidate: dict, config: Optional[GateConfig] = None) -> GateResult:
|
|
"""Decide whether a candidate may be promoted, and at what trust level."""
|
|
cfg = config or GateConfig()
|
|
reasons: list[str] = []
|
|
|
|
freq = candidate.get("frequency", 0)
|
|
sessions = _n_sessions(candidate)
|
|
impact = candidate.get("cost_impact", 0.0)
|
|
|
|
promotable = True
|
|
if freq < cfg.min_frequency:
|
|
promotable = False
|
|
reasons.append(f"frequency {freq} < min {cfg.min_frequency}")
|
|
if sessions < cfg.min_sessions:
|
|
promotable = False
|
|
reasons.append(f"sessions {sessions} < min {cfg.min_sessions}")
|
|
if impact < cfg.min_cost_impact:
|
|
promotable = False
|
|
reasons.append(f"cost_impact {impact} < min {cfg.min_cost_impact}")
|
|
|
|
dist = promotable
|
|
if cfg.dist_require_cross_flavor and not candidate.get("cross_flavor", False):
|
|
dist = False
|
|
reasons.append("not cross-flavor (required for distribution)")
|
|
if freq < cfg.dist_min_frequency:
|
|
dist = False
|
|
reasons.append(f"frequency {freq} < distribution min {cfg.dist_min_frequency}")
|
|
if impact < cfg.dist_min_cost_impact:
|
|
dist = False
|
|
reasons.append(f"cost_impact {impact} < distribution min {cfg.dist_min_cost_impact}")
|
|
|
|
return GateResult(
|
|
promotable=promotable,
|
|
distribution_ready=bool(dist),
|
|
status="approved" if dist else "provisional",
|
|
reasons=reasons,
|
|
)
|
|
|
|
|
|
def bloat_warnings(candidate: dict, existing: list[SolutionPattern]) -> list[str]:
|
|
"""Flag low-value adds against what is already catalogued (OQ6)."""
|
|
warnings: list[str] = []
|
|
cand_id = SolutionPattern.make_id(candidate["key"])
|
|
_, sig_type, locus = (candidate["key"].split(":", 2) + ["", ""])[:3]
|
|
for p in existing:
|
|
if p.id == cand_id:
|
|
warnings.append(f"duplicate of catalogued pattern {p.id}")
|
|
continue
|
|
p_parts = (p.provenance.source_key.split(":", 2) + ["", ""])[:3]
|
|
if (p_parts[1], p_parts[2]) == (sig_type, locus):
|
|
warnings.append(f"near-duplicate of {p.id} (same {sig_type}/{locus})")
|
|
return warnings
|