diff --git a/.kiro/specs/harnessx-removal/tasks.md b/.kiro/specs/harnessx-removal/tasks.md index dd5d671..4a2837e 100644 --- a/.kiro/specs/harnessx-removal/tasks.md +++ b/.kiro/specs/harnessx-removal/tasks.md @@ -22,7 +22,7 @@ and passed its fixture test. Every task is a commit gated on the full suite's ow - Fixture tests: rows from a recorded Run, reuse on an unchanged pin, recompute on a changed one, disk refusal. - _Requirements: 6.1, 6.4_ -- [ ] 3. Proposals from memory +- [x] 3. Proposals from memory - `improve/memory.py` with rules M1 (demote) and M2 (promote) and the advisory disputed-family signal; `run_round(proposals=...)`; `ousast improve --memory` (opt-in; default output byte-identical); provenance per proposal in the sidecar and journal; the train-on-test guard (qualification population, gated benchmark cases, diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index f5b790a..df40b79 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,7 +1,7 @@ { - "modules": 134, + "modules": 135, "counts": { - "load_bearing": 125, + "load_bearing": 126, "standalone_capability": 9 }, "orphaned": [], @@ -313,7 +313,8 @@ "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "cli", - "improve.evolve" + "improve.evolve", + "improve.memory" ] }, { @@ -330,6 +331,16 @@ "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "improve", + "improve.evolve", + "improve.memory" + ] + }, + { + "module": "improve.memory", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", + "importers": [ + "cli", "improve.evolve" ] }, @@ -339,7 +350,8 @@ "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "improve", - "improve.evolve" + "improve.evolve", + "improve.memory" ] }, { @@ -652,6 +664,7 @@ "cli", "improve", "improve.evolve", + "improve.memory", "model.candidates", "pair_gate" ] @@ -662,6 +675,7 @@ "reason": "ai-service-plane Req 1-4: the service plane package; runner and reconciler integration in tasks 3-4", "importers": [ "cli", + "improve.memory", "plane.egress", "plane.generate", "plane.memory", @@ -724,6 +738,7 @@ "classification": "load_bearing", "reason": "harnessx-removal Req 6.1/6.4: the memory store (file, MinIO), ingest and fact reuse; `ousast plane remember`", "importers": [ + "improve.memory", "plane.generate", "plane.tasks.remember" ] @@ -1118,6 +1133,7 @@ "hunter", "hunter_harness", "improve.evolve", + "improve.memory", "improve.validator", "pairs", "stage_processors" diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 16b56d2..6669d74 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -205,6 +205,22 @@ def main(argv: list[str] | None = None) -> int: "--profile-tolerance", type=float, default=0.0, help="allowed per-profile drop in pair_correct/Youden before rejecting" ) improve.add_argument("--min-holdout-pairs", type=int, default=5, help="profiles with fewer holdout pairs are reported, not gated") + improve.add_argument( + "--memory", + nargs="?", + const="", + default=None, + metavar="STORE", + help="also propose rule-status edits from the plane memory store (a directory, file:// or minio:// URL; " + "no value: OUSAST_MEMORY or the default store); off by default", + ) + improve.add_argument( + "--qualify-population", + action="append", + default=[], + metavar="NAME", + help="population a memory proposal will be qualified on; its rows never propose (repeatable)", + ) pairs = subparsers.add_parser( "pairs", @@ -367,6 +383,9 @@ def main(argv: list[str] | None = None) -> int: pair_catalog=None if args.no_pair_gate else args.pair_catalog, profile_tolerance=args.profile_tolerance, min_holdout_pairs=args.min_holdout_pairs, + memory=args.memory, + qualify_populations=tuple(args.qualify_population), + guard_catalog=args.pair_catalog, ) if args.command == "pairs": return _pairs( @@ -929,6 +948,9 @@ def _improve( pair_catalog: Path | None = None, profile_tolerance: float = 0.0, min_holdout_pairs: int = 5, + memory: str | None = None, + qualify_populations: tuple[str, ...] = (), + guard_catalog: Path | None = None, ) -> int: if not manifest_path.exists() or not manifest_path.is_file(): raise SystemExit(f"benchmark manifest is not a file: {manifest_path}") @@ -940,6 +962,7 @@ def _improve( manifest = load_benchmark_manifest(manifest_path) target = resolve_benchmark_source(manifest_path, manifest) policy = load_policy() + session = _memory_session(memory, manifest, target, ruleset_dir, qualify_populations, guard_catalog) if memory is not None else None # The loop writes its accepted ledger where `scan`/`benchmark` read it, so a # subsequent scan of this target automatically picks up the improved ruleset. @@ -965,11 +988,22 @@ def _improve( pair_cases=pair_cases, profile_tolerance=profile_tolerance, min_holdout_pairs=min_holdout_pairs, + proposals=session.proposals(ledger_path, journal_path) if session else (), ) _print_improve_outcomes(outcomes, manifest_path, target, ledger_path, dry_run=dry_run) + if session is not None: + session.finish(outcomes, journal_path, dry_run=dry_run) return 0 +def _memory_session( + memory: str, manifest: object, target: Path, ruleset_dir: Path, populations: tuple[str, ...], catalog: Path | None +) -> Any: + from .improve.memory import MemorySession # lazy: the default `improve` path never touches the store + + return MemorySession.open(memory, manifest, target, ruleset_dir, populations, catalog) + + def _pairs( catalog: Path, slice_name: str, diff --git a/src/openultrasast/improve/evolve.py b/src/openultrasast/improve/evolve.py index b8ffb6e..aafdfc9 100644 --- a/src/openultrasast/improve/evolve.py +++ b/src/openultrasast/improve/evolve.py @@ -5,8 +5,8 @@ edits (auto-shadow precision-draggers) -> validate -> replay smoke -> re-benchma hard acceptance gate (recall >= floor AND FP < ceiling AND project score not regressed AND no matched-finding regression) -> accept (persist the ledger) or revert byte-for-byte. -The edit *proposer* here is the benchmark's per-rule signal; the LLM ``MetaAgent.evolve`` -proposer plugs into this same validated/gated machinery when the HarnessX extra is present. +The edit *proposers* are the benchmark's per-rule signal and, opt-in, the plane's memory store +(:mod:`.memory`, ``run_round(proposals=...)``); both go through this same validator and gate. The project score is the optimization reward; the recall/FP gate is a separate hard feasibility constraint that is never folded into the reward. """ @@ -27,6 +27,7 @@ from ..rank import rank_targets from ..ruleset import DEFAULT_RULESET_DIR, PatternRule, load_ruleset, read_rule_ledger, write_rule_ledger from ..scoring import build_score_artifact from .journal import append_round, load_journal, next_round_index, reverted_edit_keys +from .memory import MemoryProposal, applicable, proposal_id_of, write_sidecar from .validator import EvolveValidator, RuleStatusEdit, StrictValidationError, edits_to_ledger @@ -159,6 +160,7 @@ def run_round( pair_cases: Sequence[object] | None = None, profile_tolerance: float = 0.0, min_holdout_pairs: int = 5, + proposals: Sequence[MemoryProposal] = (), ) -> RoundOutcome: policy = policy if policy is not None else load_policy() rounds = load_journal(journal_path) @@ -173,6 +175,14 @@ def run_round( ruleset_by_id = {rule.rule_id: rule for rule in before_rules} edits = propose_status_edits(before.result.metrics.per_rule, ruleset_by_id, current_ledger, blocked) + if proposals: + # harnessx-removal Req 6.2: memory proposals join after the benchmark's edits; the first edit per key wins. + taken = {edit.key() for edit in edits} + for proposal in applicable(proposals, ruleset_by_id, current_ledger, rounds): + if proposal.edit.key() not in taken: + taken.add(proposal.edit.key()) + edits.append(proposal.edit) + write_sidecar(journal_path, round_index, [p for p in proposals if p.edit in edits]) refusals: list[dict[str, object]] = [] if not edits: return _outcome(round_index, accepted=False, reason="no_proposals", edits=[], before=before, after=before, degradations=refusals) @@ -260,6 +270,7 @@ def run_improvement( pair_cases: Sequence[object] | None = None, profile_tolerance: float = 0.0, min_holdout_pairs: int = 5, + proposals: Sequence[MemoryProposal] = (), ) -> list[RoundOutcome]: """Run improvement rounds until convergence (no new proposals) or ``max_rounds``.""" outcomes: list[RoundOutcome] = [] @@ -276,6 +287,7 @@ def run_improvement( pair_cases=pair_cases, profile_tolerance=profile_tolerance, min_holdout_pairs=min_holdout_pairs, + proposals=proposals, ) outcomes.append(outcome) if outcome.reason == "no_proposals": @@ -382,10 +394,7 @@ def _record( "reason": reason or outcome, "profile_regressions": list(profile_regressions or []), "profiles_under_minimum": list(profiles_under_minimum or []), - "edits": [ - {"key": e.key(), "lever": e.lever, "rule_id": e.rule_id, "from": e.from_status, "to": e.to_status, "rationale": e.rationale} - for e in edits - ], + "edits": [_journal_edit(e) for e in edits], "recall_before": round(before.recall, 4), "recall_after": round(after.recall, 4), "fp_before": round(before.fp_rate, 4), @@ -394,3 +403,14 @@ def _record( "score_after": after.score, }, ) + + +def _journal_edit(edit: RuleStatusEdit) -> dict[str, object]: + entry: dict[str, object] = { + "key": edit.key(), "lever": edit.lever, "rule_id": edit.rule_id, "from": edit.from_status, "to": edit.to_status, + "rationale": edit.rationale, + } # fmt: skip + proposal_id = proposal_id_of(edit) + if proposal_id is not None: + entry["evidence"] = proposal_id # only a memory edit carries it; every other entry is unchanged byte for byte + return entry diff --git a/src/openultrasast/improve/memory.py b/src/openultrasast/improve/memory.py new file mode 100644 index 0000000..aea27f5 --- /dev/null +++ b/src/openultrasast/improve/memory.py @@ -0,0 +1,513 @@ +"""Rule-status proposals from the plane's memory store (harnessx-removal Req 6.2/6.4, design section 5). + +The store (:mod:`..plane.memory`) holds what plane runs learned per repository and pin. Two deterministic rules turn +its ``alert`` and ``verdict`` rows into :class:`RuleStatusEdit` proposals that :func:`.evolve.run_round` validates +and gates exactly like the benchmark's own edits; a third reads dispute rates as advisory routing input only. + +Join key: an ``alert`` and a ``verdict`` match when repo, pin and ``path::function`` are equal. A ``disputed`` +final counts as no evidence in either direction. + +- **M1, repeated false alerts** (``enabled -> shadow``): at least :data:`M1_MIN_FALSE` alerts of the rule are false + -- at a ``rejected`` candidate, or on a ``fixed`` pin inside the declared fix range (the alert row says + ``in_fix_range: true``, or its ``path::function`` is a ``site_match`` site of the same repository) -- across at + least :data:`M1_MIN_REPOS` repositories, and no alert of the rule sits at an ``agreed`` or ``site_match`` + candidate. +- **M2, missed declared sites** (``shadow -> enabled``): the shadow rule's alerts hit at least :data:`M2_MIN_HITS` + ``agreed`` candidates with ``site_match`` true across at least :data:`M2_MIN_REPOS` repositories, no enabled rule + alerts at those candidates, the rule has no ``fixed``-pin alert at the same function, and at most + :data:`M2_MAX_REJECTED` alert at a ``rejected`` candidate. +- **M3, disputed families**: families with at least :data:`M3_MIN_CANDIDATES` candidates and a dispute rate of at + least :data:`M3_MIN_DISPUTE_RATE`. Advisory (``signals.json``); never an edit, never shown to the validator. + +**Train-on-test guard** (:class:`Guard`), applied before any rule sees a row: a row is dropped when its population +is one the proposal will be qualified on, its (repo, pin) is a case of the gated benchmark manifest, or its +repository is a holdout pair's repository. The pair catalog records a holdout pair's *fix* commit, not the +vulnerable pin (its parent), so the holdout clause matches by repository, a superset of (repo, pin). This is the +rule the closed-loop leak broke, where a holdout pair taught the shape that then "recovered" it. + +**Provenance**: each proposal's rationale is ``memory:``, ``proposal_id`` being the sha256 of (rule, +edit key, sorted evidence row ids, thresholds). The sidecar ``memory_proposals.jsonl`` next to the journal carries +the evidence rows (id, repo, pin, run, task, object key and version -- MinIO's version id, a file's content hash), +the counts that met each threshold, what the guard excluded and the store's index digest. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass, field +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from ..plane.memory import MemoryStore, Record, repo_key, row_id, rows_digest +from ..ruleset import PatternRule +from .validator import RuleStatusEdit + +M1_MIN_FALSE = 3 +M1_MIN_REPOS = 2 +M2_MIN_HITS = 2 +M2_MIN_REPOS = 2 +M2_MAX_REJECTED = 1 +M3_MIN_CANDIDATES = 5 +M3_MIN_DISPUTE_RATE = 0.3 +THRESHOLDS: dict[str, dict[str, float]] = { + "M1": {"min_false_alerts": M1_MIN_FALSE, "min_repos": M1_MIN_REPOS, "max_agreed_hits": 0}, + "M2": {"min_site_hits": M2_MIN_HITS, "min_repos": M2_MIN_REPOS, "max_rejected": M2_MAX_REJECTED, "max_fixed_pin_same_function": 0}, + "M3": {"min_candidates": M3_MIN_CANDIDATES, "min_dispute_rate": M3_MIN_DISPUTE_RATE}, +} +RATIONALE_PREFIX = "memory:" +SIDECAR = "memory_proposals.jsonl" +SIGNALS = "signals.json" + + +@dataclass(frozen=True) +class MemoryProposal: + edit: RuleStatusEdit + provenance: dict[str, Any] + + @property + def proposal_id(self) -> str: + return str(self.provenance["proposal_id"]) + + +def proposal_id_of(edit: object) -> str | None: + """The proposal id a memory edit carries in its rationale; None for any other edit.""" + rationale = getattr(edit, "rationale", "") + return rationale[len(RATIONALE_PREFIX) :] if isinstance(rationale, str) and rationale.startswith(RATIONALE_PREFIX) else None + + +# --- the train-on-test guard --------------------------------------------------------------------------------------- + + +def _tail(repo: str) -> str: + """``owner/name`` of a repository, lower-cased: the pair catalog names repositories without a host.""" + return "/".join(repo_key(repo).lower().split("/")[-2:]) + + +@dataclass(frozen=True) +class Guard: + populations: frozenset[str] = frozenset() + gated_cases: frozenset[tuple[str, str]] = frozenset() # (repo_key, pin) of the benchmark being gated + holdout_repos: frozenset[str] = frozenset() # owner/name of every holdout pair + + @classmethod + def build( + cls, populations: Iterable[str] = (), gated_cases: Iterable[tuple[str, str]] = (), holdout_pairs: Iterable[object] = () + ) -> Guard: + repos = {_tail(str(getattr(p, "repo", ""))) for p in holdout_pairs if getattr(p, "repo", "")} + return cls(frozenset(populations), frozenset((repo_key(r), p) for r, p in gated_cases), frozenset(repos)) + + def reason(self, row: Mapping[str, Any]) -> str | None: + if row.get("population") in self.populations: + return "population" + if (row.get("repo"), row.get("pin")) in self.gated_cases: + return "gated_case" + if _tail(str(row.get("repo"))) in self.holdout_repos: + return "holdout_pair" + return None + + def apply(self, records: Iterable[Record]) -> tuple[list[Record], dict[str, Any]]: + """(kept records, what was excluded: row count, and counts per population / repo@pin per clause).""" + kept: list[Record] = [] + excluded: dict[str, Any] = {"rows": 0, "population": {}, "gated_case": {}, "holdout_pair": {}} + for record in records: + why = self.reason(record.row) + if why is None: + kept.append(record) + continue + excluded["rows"] += 1 + label = str(record.row.get("population")) if why == "population" else f"{record.row.get('repo')}@{record.row.get('pin')}" + excluded[why][label] = excluded[why].get(label, 0) + 1 + for clause in ("population", "gated_case", "holdout_pair"): + excluded[clause] = dict(sorted(excluded[clause].items())) + return kept, excluded + + +# --- the rules ----------------------------------------------------------------------------------------------------- + + +def _cand(row: Mapping[str, Any]) -> str: + return f"{row.get('path')}::{row.get('function')}" + + +def _status(rule_id: str, ruleset_by_id: Mapping[str, PatternRule], ledger: Mapping[str, dict[str, object]]) -> str | None: + rule = ruleset_by_id.get(rule_id) + return None if rule is None else str(ledger.get(rule_id, {}).get("status", rule.status)) + + +@dataclass +class _View: + verdicts: dict[tuple[str, str, str], Record] = field(default_factory=dict) # (repo, pin, candidate) + sites: set[tuple[str, str]] = field(default_factory=set) # (repo, candidate) with site_match true + alerts: dict[str, list[Record]] = field(default_factory=dict) # rule_id -> alerts + + @classmethod + def of(cls, records: Sequence[Record]) -> _View: + view = cls() + for rec in records: + row = rec.row + if row["kind"] == "verdict": + view.verdicts[(row["repo"], row["pin"], str(row.get("candidate")))] = rec + if row.get("site_match") is True: + view.sites.add((row["repo"], str(row.get("candidate")))) + elif row["kind"] == "alert" and row.get("rule_id"): + view.alerts.setdefault(str(row["rule_id"]), []).append(rec) + return view + + def verdict(self, alert: Mapping[str, Any]) -> Record | None: + return self.verdicts.get((alert["repo"], alert["pin"], _cand(alert))) + + def fixed_in_range(self, alert: Mapping[str, Any]) -> bool: + if alert.get("pin_role") != "fixed": + return False + flag = alert.get("in_fix_range") + return bool(flag) if flag is not None else (alert["repo"], _cand(alert)) in self.sites + + +def _m1(rule_id: str, view: _View) -> tuple[list[Record], dict[str, int]] | None: + evidence: list[Record] = [] + repos: set[str] = set() + false_alerts = 0 + for alert in view.alerts.get(rule_id, []): + verdict = view.verdict(alert.row) + final = verdict.row.get("final") if verdict else None + if final == "agreed" or (verdict is not None and verdict.row.get("site_match") is True): + return None # the rule caught something the plane agreed on: never demote it on this evidence + if final == "rejected" or view.fixed_in_range(alert.row): + false_alerts += 1 + repos.add(alert.row["repo"]) + evidence.extend([alert, verdict] if verdict is not None and final == "rejected" else [alert]) + if false_alerts < M1_MIN_FALSE or len(repos) < M1_MIN_REPOS: + return None + return evidence, {"false_alerts": false_alerts, "repos": len(repos), "agreed_hits": 0} + + +def _m2(rule_id: str, view: _View, enabled: set[str]) -> tuple[list[Record], dict[str, int]] | None: + alerts = view.alerts.get(rule_id, []) + hits: dict[tuple[str, str, str], tuple[Record, Record]] = {} + rejected = 0 + for alert in alerts: + verdict = view.verdict(alert.row) + if verdict is None or alert.row.get("pin_role") == "fixed": + continue + if verdict.row.get("final") == "rejected": + rejected += 1 + elif verdict.row.get("final") == "agreed" and verdict.row.get("site_match") is True: + hits[(alert.row["repo"], alert.row["pin"], _cand(alert.row))] = (alert, verdict) + if not hits: + return None + caught = { + (a.row["repo"], a.row["pin"], _cand(a.row)) for rid in enabled if rid != rule_id for a in view.alerts.get(rid, []) + } # fmt: skip + missed = {k: v for k, v in hits.items() if k not in caught} + fixed_same = sum( + 1 for a in alerts if a.row.get("pin_role") == "fixed" and any((a.row["repo"], _cand(a.row)) == (r, c) for r, _, c in missed) + ) + repos = {repo for repo, _, _ in missed} + if len(missed) < M2_MIN_HITS or len(repos) < M2_MIN_REPOS or fixed_same > 0 or rejected > M2_MAX_REJECTED: + return None + evidence = [rec for key in sorted(missed) for rec in missed[key]] + return evidence, {"site_hits": len(missed), "repos": len(repos), "rejected": rejected, "fixed_pin_same_function": 0} + + +def _evidence(records: Sequence[Record]) -> list[dict[str, Any]]: + unique = {rec.row["id"]: rec for rec in records} + return [ + {"id": i, "kind": r.row["kind"], "repo": r.row["repo"], "pin": r.row["pin"], "run": r.row["run"], "task": r.row["task"], + "key": r.key, "version": r.version} + for i, r in sorted(unique.items()) + ] # fmt: skip + + +def evidence_digest(evidence_ids: Iterable[str]) -> str: + return hashlib.sha256("\n".join(sorted(evidence_ids)).encode("utf-8")).hexdigest() + + +def _proposal( + rule: str, edit_key: str, rule_id: str, records: Sequence[Record], counts: dict[str, int], excluded: dict[str, Any], index: str +) -> dict[str, Any]: + evidence = _evidence(records) + ids = [e["id"] for e in evidence] + seed = json.dumps([rule, edit_key, ids, THRESHOLDS[rule]], sort_keys=True, separators=(",", ":")) + return { + "proposal_id": hashlib.sha256(seed.encode("utf-8")).hexdigest(), + "rule": rule, + "edit_key": edit_key, + "rule_id": rule_id, + "evidence": evidence, + "evidence_digest": evidence_digest(ids), + "counts": counts, + "thresholds": THRESHOLDS[rule], + "excluded": excluded, + "index_digest": index, + } + + +def propose_from_memory( + rows: Sequence[Record], + ruleset_by_id: Mapping[str, PatternRule], + current_ledger: Mapping[str, dict[str, object]], + journal_rounds: Sequence[Mapping[str, object]], + *, + excluded: Guard, + index_digest: str = "", +) -> list[MemoryProposal]: + """M1 and M2 over the rows the guard keeps, minus proposals already reverted on the same evidence.""" + kept, dropped = excluded.apply(rows) + view = _View.of(kept) + statuses = {rid: _status(rid, ruleset_by_id, current_ledger) for rid in ruleset_by_id} + enabled = {rid for rid, status in statuses.items() if status == "enabled"} + proposals: list[MemoryProposal] = [] + for rule_id in sorted(view.alerts): + status = statuses.get(rule_id) + if status == "enabled" and (m1 := _m1(rule_id, view)): + name, found, to_status = "M1", m1, "shadow" + elif status == "shadow" and (m2 := _m2(rule_id, view, enabled)): + name, found, to_status = "M2", m2, "enabled" + else: + continue # unknown to the ruleset (the loop cannot add rules), disabled, or no rule fired + key = RuleStatusEdit(rule_id, status, to_status).key() + provenance = _proposal(name, key, rule_id, found[0], found[1], dropped, index_digest) + edit = RuleStatusEdit(rule_id, status, to_status, rationale=RATIONALE_PREFIX + provenance["proposal_id"]) + proposals.append(MemoryProposal(edit, provenance)) + return applicable(proposals, ruleset_by_id, current_ledger, journal_rounds) + + +def applicable( + proposals: Sequence[MemoryProposal], + ruleset_by_id: Mapping[str, PatternRule], + current_ledger: Mapping[str, dict[str, object]], + journal_rounds: Sequence[Mapping[str, object]], +) -> list[MemoryProposal]: + """Proposals whose rule still has the ``from`` status and whose evidence no reverted round already tried.""" + tried: set[tuple[str, str]] = set() + for entry in journal_rounds: + edits = entry.get("edits") + if entry.get("outcome") == "reverted" and isinstance(edits, list): + tried |= {(str(e.get("key")), str(e.get("evidence"))) for e in edits if isinstance(e, dict) and e.get("evidence")} + return [ + p for p in proposals + if _status(p.edit.rule_id, ruleset_by_id, current_ledger) == p.edit.from_status and (p.edit.key(), p.proposal_id) not in tried + ] # fmt: skip + + +def disputed_families(rows: Sequence[Record], guard: Guard) -> list[dict[str, Any]]: + """M3: families disputed often enough to route differently. Advisory only: never an edit.""" + kept, _ = guard.apply(rows) + per: dict[str, list[int]] = {} + for rec in kept: + if rec.row["kind"] == "verdict" and rec.row.get("family"): + tally = per.setdefault(str(rec.row["family"]), [0, 0]) + tally[0] += 1 + tally[1] += rec.row.get("final") == "disputed" + return [ + {"family": fam, "candidates": n, "disputed": d, "dispute_rate": round(d / n, 4), "advice": "schedule pass c up front"} + for fam, (n, d) in sorted(per.items()) + if n >= M3_MIN_CANDIDATES and d / n >= M3_MIN_DISPUTE_RATE + ] + + +# --- the store, the sidecar and the outcomes ----------------------------------------------------------------------- + + +def index_digest(store: MemoryStore) -> str: + return rows_digest(store.index()) + + +def write_sidecar(journal_path: Path, round_index: int, proposals: Sequence[MemoryProposal]) -> None: + """Append each proposal once (by id) to ``memory_proposals.jsonl`` next to the journal.""" + if not proposals: + return + path = journal_path.with_name(SIDECAR) + seen = {json.loads(line)["proposal_id"] for line in path.read_text().splitlines() if line.strip()} if path.is_file() else set() + new = [p for p in proposals if p.proposal_id not in seen] + if not new: + return + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a") as sink: + for p in new: + sink.write(json.dumps({**p.provenance, "round": round_index}, sort_keys=True, separators=(",", ":")) + "\n") + + +def read_sidecar(journal_path: Path) -> dict[str, dict[str, Any]]: + path = journal_path.with_name(SIDECAR) + if not path.is_file(): + return {} + return {row["proposal_id"]: row for row in (json.loads(line) for line in path.read_text().splitlines() if line.strip())} + + +def write_signals(directory: Path, signals: Sequence[Mapping[str, Any]]) -> Path: + path = directory / SIGNALS + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"advisory": True, "disputed_families": list(signals), "thresholds": THRESHOLDS["M3"]}, indent=2) + "\n") + return path + + +def outcome_of(reason: str, accepted: bool) -> str | None: + if accepted: + return "accepted" + if reason == "no_proposals": + return None + if reason.startswith(("validation_failed", "replay_failed")): + return "rejected" + return "reverted" + + +def outcome_rows( + edits: Sequence[RuleStatusEdit], + outcome: str, + round_index: int, + reason: str, + proposals: Mapping[str, MemoryProposal], + gate_run: str, +) -> list[dict[str, Any]]: + """One ``proposal_outcome`` row per (repository, pin) of each memory edit's evidence.""" + rows: list[dict[str, Any]] = [] + for edit in edits: + pid = proposal_id_of(edit) + proposal = proposals.get(pid or "") + if proposal is None: + continue + by_place: dict[tuple[str, str], dict[str, Any]] = {} + for ev in proposal.provenance["evidence"]: + by_place.setdefault((ev["repo"], ev["pin"]), ev) + for (repo, pin), _ in sorted(by_place.items()): + task = f"round-{round_index}" + rows.append( + { + "id": row_id("proposal_outcome", gate_run, task, f"{pid}:{repo}:{pin}"), "kind": "proposal_outcome", + "repo": repo, "pin": pin, "run": gate_run, "task": task, "population": "improve-loop", "split": "loop", + "image": "host", "edit_key": edit.key(), "proposal_id": pid, + "evidence_digest": proposal.provenance["evidence_digest"], "outcome": outcome, "reason": reason, + "gate_run": gate_run, + } + ) # fmt: skip + return rows + + +def gate_run_name(manifest_name: str) -> str: + return f"improve:{manifest_name}:{datetime.now(UTC).strftime('%Y%m%dT%H%M%SZ')}" + + +# --- the `ousast improve --memory` session ------------------------------------------------------------------------- + + +def benchmark_cases(manifest: object) -> frozenset[tuple[str, str]]: + """(repo, pin) of the gated benchmark's cases. Local sources (the only kind `resolve_benchmark_source` accepts) + name no repository pin, so they exclude nothing here; a pinned source would list its case.""" + source = getattr(manifest, "source", None) + repo, pin = getattr(source, "repo", ""), getattr(source, "pin", "") + return frozenset({(repo_key(repo), pin)}) if repo and pin else frozenset() + + +def store_spec(value: str) -> str | None: + """``--memory`` value -> ``open_store`` spec: empty uses OUSAST_MEMORY or the default, a bare path is a FileStore.""" + if not value: + return None + return value if "://" in value else f"file://{Path(value).resolve()}" + + +@dataclass +class MemorySession: + """One `improve --memory` invocation: the store read once, proposals per round, outcomes written back.""" + + store: MemoryStore + guard: Guard + records: list[Record] + ruleset_dir: Path + gate_run: str + made: dict[str, MemoryProposal] = field(default_factory=dict) + + @classmethod + def open( + cls, spec: str, manifest: object, target: Path, ruleset_dir: Path, populations: Sequence[str], catalog: Path | None + ) -> MemorySession: + from ..pairs import load_pair_catalog, select_split + from ..plane.memory import FileStore, open_store + + store = open_store(store_spec(spec)) + if isinstance(store, FileStore) and not (store.root / "index.jsonl").is_file(): + raise SystemExit(f"memory store {store.describe()} has no index.jsonl: an unread store would propose nothing") + if catalog is None or not catalog.is_file(): + raise SystemExit(f"--memory needs the pair catalog for the holdout guard: {catalog} is not a file") + holdout = select_split(load_pair_catalog(catalog), "holdout") + guard = Guard.build(populations, benchmark_cases(manifest), holdout) + records = store.rows() + print( + f"memory={store.describe()} index_entries={len(store.index())} rows={len(records)} " + f"guard: populations={sorted(guard.populations) or '-'} gated_cases={len(guard.gated_cases)} " + f"holdout_repos={len(guard.holdout_repos)}" + ) + return cls(store, guard, records, ruleset_dir, gate_run_name(str(getattr(manifest, "name", "manifest")))) + + def proposals(self, ledger_path: Path, journal_path: Path) -> list[MemoryProposal]: + from ..ruleset import load_ruleset, read_rule_ledger + from .journal import load_journal + + rules = load_ruleset(self.ruleset_dir, ledger_path if ledger_path.is_file() else None) + found = propose_from_memory( + self.records, + {r.rule_id: r for r in rules}, + read_rule_ledger(ledger_path), + load_journal(journal_path), + excluded=self.guard, + index_digest=index_digest(self.store), + ) + kept, excluded = self.guard.apply(self.records) + kinds: dict[str, int] = {} + for rec in kept: + kinds[rec.row["kind"]] = kinds.get(rec.row["kind"], 0) + 1 + print(f"memory_rows kept={len(kept)} by_kind={json.dumps(dict(sorted(kinds.items())))}") + print(f"memory_excluded={json.dumps(excluded, sort_keys=True)}") + for p in found: + print(f"memory_proposal {p.edit.key()} rule={p.provenance['rule']} id={p.proposal_id} evidence={len(p.provenance['evidence'])}") + print(f"memory_proposals={len(found)}") + self.made.update({p.proposal_id: p for p in found}) + return found + + def finish(self, outcomes: Sequence[Any], journal_path: Path, *, dry_run: bool) -> None: + signals = disputed_families(self.records, self.guard) + path = write_signals(journal_path.parent, signals) + print(f"memory_signals disputed_families={[s['family'] for s in signals] or '-'} ({path.name}, advisory)") + rows: list[dict[str, Any]] = [] + for o in outcomes: + outcome = outcome_of(o.reason, o.accepted) + if outcome is not None: + rows += outcome_rows(o.edits, outcome, o.round, o.reason, self.made, self.gate_run) + if dry_run: + print(f"memory_outcomes={len(rows)} (dry run: not written to the store)") + return + if rows: + self.store.put_rows(rows) + print(f"memory_outcomes={len(rows)} written to {self.store.describe()}") + + +__all__ = [ + "Guard", + "MemorySession", + "benchmark_cases", + "store_spec", + "M1_MIN_FALSE", + "M1_MIN_REPOS", + "M2_MAX_REJECTED", + "M2_MIN_HITS", + "M2_MIN_REPOS", + "M3_MIN_CANDIDATES", + "M3_MIN_DISPUTE_RATE", + "MemoryProposal", + "THRESHOLDS", + "applicable", + "disputed_families", + "evidence_digest", + "gate_run_name", + "index_digest", + "outcome_of", + "outcome_rows", + "propose_from_memory", + "proposal_id_of", + "read_sidecar", + "write_sidecar", + "write_signals", +] diff --git a/tests/test_improve_memory.py b/tests/test_improve_memory.py new file mode 100644 index 0000000..2a458c5 --- /dev/null +++ b/tests/test_improve_memory.py @@ -0,0 +1,302 @@ +"""Proposals from plane memory (harnessx-removal task 3, design section 5): rules M1/M2, the advisory M3 signal, +the train-on-test guard, provenance, and the unchanged default `improve`.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from openultrasast.benchmark import load_benchmark_manifest +from openultrasast.cli import main +from openultrasast.improve import EvolveValidator, RuleStatusEdit, run_round +from openultrasast.improve.memory import ( + SIDECAR, + Guard, + MemoryProposal, + disputed_families, + outcome_of, + outcome_rows, + propose_from_memory, + read_sidecar, +) +from openultrasast.pairs import PairCase +from openultrasast.plane.memory import FileStore, Record, row_id +from openultrasast.ruleset import PatternRule, read_rule_ledger, write_ruleset + +P = {name: ch * 40 for name, ch in (("a", "a"), ("b", "b"), ("c", "c"), ("f", "f"))} +NO_GUARD = Guard() + + +def _rule(rule_id: str, pattern: str = r"\bexec\s*\(", status: str = "enabled") -> PatternRule: + return PatternRule( + rule_id=rule_id, title=rule_id, languages=("python",), cwe="CWE-95", tags=("injection",), pattern=pattern, status=status + ) + + +def _row(kind: str, repo: str, pin: str, subject: str, population: str = "dev", **fields: Any) -> dict[str, Any]: + return { + "id": row_id(kind, "run-1", f"{repo}-task", subject), "kind": kind, "repo": f"github.com/o/{repo}", "pin": pin, + "run": "run-1", "task": f"{repo}-task", "population": population, "split": "validation", "image": "sha256:x", **fields, + } # fmt: skip + + +def verdict(repo: str, pin: str, fn: str, final: str, site_match: bool | None = None, **kw: Any) -> dict[str, Any]: + cand = f"app.py::{fn}" + return _row("verdict", repo, pin, cand, candidate=cand, family="injection", final=final, site_match=site_match, **kw) + + +def alert(rule: str, repo: str, pin: str, fn: str, role: str = "vulnerable", status: str = "enabled", **kw: Any) -> dict[str, Any]: + fields = {"rule_id": rule, "rule_status": status, "path": "app.py", "line": 3, "function": fn, "pin_role": role} + return _row("alert", repo, pin, f"{rule}:{fn}:{role}", **fields, **kw) + + +def recs(rows: list[dict[str, Any]]) -> list[Record]: + return [Record(r, f"repos/{r['repo']}/{r['pin']}.jsonl", None) for r in rows] + + +def propose(rows: list[dict[str, Any]], rules: list[PatternRule], guard: Guard = NO_GUARD, ledger: Any = None) -> list[MemoryProposal]: + return propose_from_memory(recs(rows), {r.rule_id: r for r in rules}, ledger or {}, [], excluded=guard) + + +def false_alerts(rule: str = "noisy", repos: tuple[str, ...] = ("r1", "r1", "r2"), population: str = "dev") -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for i, repo in enumerate(repos): + rows.append(verdict(repo, P["a"], f"f{i}", "rejected", population=population)) + rows.append(alert(rule, repo, P["a"], f"f{i}", population=population)) + return rows + + +# ---- M1: repeated false alerts ------------------------------------------------------------------------------------- + + +def test_m1_demotes_a_rule_with_false_alerts_across_repositories() -> None: + [proposal] = propose(false_alerts(), [_rule("noisy")]) + assert (proposal.edit.from_status, proposal.edit.to_status) == ("enabled", "shadow") + assert proposal.edit.rationale == f"memory:{proposal.proposal_id}" + assert proposal.provenance["rule"] == "M1" + assert proposal.provenance["counts"] == {"false_alerts": 3, "repos": 2, "agreed_hits": 0} + assert len(proposal.provenance["evidence"]) == 6 # three alerts and the three rejected verdicts they sit at + + +@pytest.mark.parametrize("repos", [("r1", "r2"), ("r1", "r1", "r1")], ids=["two-alerts", "one-repository"]) +def test_m1_stays_silent_below_threshold(repos: tuple[str, ...]) -> None: + assert propose(false_alerts(repos=repos), [_rule("noisy")]) == [] + + +def test_m1_never_demotes_a_rule_with_an_agreed_hit() -> None: + rows = false_alerts() + [verdict("r3", P["b"], "hit", "agreed"), alert("noisy", "r3", P["b"], "hit")] + assert propose(rows, [_rule("noisy")]) == [] + site = false_alerts() + [verdict("r3", P["b"], "site", "rejected", site_match=True), alert("noisy", "r3", P["b"], "site")] + assert propose(site, [_rule("noisy")]) == [] + + +def test_m1_counts_fixed_pin_alerts_in_the_fix_range_and_ignores_disputes() -> None: + rows = [ + verdict("r1", P["a"], "sink", "agreed", site_match=True), + verdict("r2", P["b"], "sink", "agreed", site_match=True), + alert("noisy", "r1", P["f"], "sink", role="fixed"), # the declared site's function on the fixed pin + alert("noisy", "r2", P["f"], "sink", role="fixed"), + alert("noisy", "r2", P["c"], "other", role="fixed", in_fix_range=True), + verdict("r1", P["a"], "maybe", "disputed"), + alert("noisy", "r1", P["a"], "maybe"), # disputed: evidence in neither direction + ] + [proposal] = propose(rows, [_rule("noisy")]) + assert proposal.provenance["counts"]["false_alerts"] == 3 + + +def test_m1_skips_rules_the_ruleset_does_not_know_or_that_are_not_enabled() -> None: + assert propose(false_alerts(rule="ghost"), [_rule("noisy")]) == [] + assert propose(false_alerts(), [_rule("noisy")], ledger={"noisy": {"status": "shadow"}}) == [] + + +# ---- M2: missed declared sites ------------------------------------------------------------------------------------- + + +def site_hits(rule: str = "quiet") -> list[dict[str, Any]]: + return [ + verdict("r1", P["a"], "s1", "agreed", site_match=True), + verdict("r2", P["b"], "s2", "agreed", site_match=True), + alert(rule, "r1", P["a"], "s1", status="shadow"), + alert(rule, "r2", P["b"], "s2", status="shadow"), + ] + + +def test_m2_promotes_a_shadow_rule_that_hits_declared_sites_nothing_else_catches() -> None: + [proposal] = propose(site_hits(), [_rule("quiet", status="shadow"), _rule("other")]) + assert (proposal.edit.from_status, proposal.edit.to_status, proposal.provenance["rule"]) == ("shadow", "enabled", "M2") + assert proposal.provenance["counts"]["site_hits"] == 2 + + +def test_m2_stays_silent_below_threshold_or_when_already_caught() -> None: + rules = [_rule("quiet", status="shadow"), _rule("other")] + assert propose(site_hits()[:3], rules) == [] # one hit + assert propose([*site_hits(), alert("other", "r1", P["a"], "s1")], rules) == [] # an enabled rule catches s1 + + +def test_m2_stays_silent_on_a_fixed_pin_alert_or_rejected_candidates() -> None: + rules = [_rule("quiet", status="shadow")] + assert propose([*site_hits(), alert("quiet", "r1", P["f"], "s1", role="fixed", status="shadow")], rules) == [] + rejected = [ + verdict("r3", P["c"], "x1", "rejected"), + verdict("r3", P["c"], "x2", "rejected"), + alert("quiet", "r3", P["c"], "x1", status="shadow"), + alert("quiet", "r3", P["c"], "x2", status="shadow"), + ] + assert propose([*site_hits(), *rejected[::2]], rules) != [] # one rejected alert is tolerated + assert propose([*site_hits(), *rejected], rules) == [] + + +# ---- M3: advisory only --------------------------------------------------------------------------------------------- + + +def test_disputed_families_are_advisory_and_never_reach_the_validator(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + rows = [verdict("r1", P["a"], f"d{i}", "disputed" if i < 2 else "agreed") for i in range(5)] + [signal] = disputed_families(recs(rows), NO_GUARD) + assert signal["family"] == "injection" and signal["dispute_rate"] == 0.4 + assert disputed_families(recs(rows[1:]), NO_GUARD) == [] # four candidates: under the minimum + assert propose(rows, [_rule("noisy")]) == [] # a dispute is no proposal + + seen: list[object] = [] + original = EvolveValidator.validate + monkeypatch.setattr(EvolveValidator, "validate", lambda self, edit, *a: (seen.append(edit), original(self, edit, *a))[1]) + repo, rules_dir, manifest = _scaffold(tmp_path) + [proposal] = propose(false_alerts(), [_rule("noisy")]) + run_round( + repo, manifest, ledger_path=tmp_path / "l.json", journal_path=tmp_path / "j.json", ruleset_dir=rules_dir, proposals=[proposal] + ) + assert seen and all(isinstance(edit, RuleStatusEdit) for edit in seen) + + +# ---- the train-on-test guard --------------------------------------------------------------------------------------- + + +def test_guard_drops_the_qualification_population_and_reports_it() -> None: + rows = false_alerts(population="independent-v2/validation") + false_alerts(repos=("r4",), population="dev") + guard = Guard.build(populations=["independent-v2/validation"]) + kept, excluded = guard.apply(recs(rows)) + assert len(kept) == 2 and excluded["rows"] == 6 and excluded["population"] == {"independent-v2/validation": 6} + assert propose(rows, [_rule("noisy")], guard) == [] # the only evidence is excluded: no proposal + assert propose(rows, [_rule("noisy")]) != [] # the same rows without the guard would propose + + +def test_guard_drops_holdout_pair_repositories_and_gated_cases() -> None: + holdout = PairCase( + name="h", slice="s", language="python", origin="o", vuln_file=Path("v"), fixed_file=Path("f"), relpath="app.py", + expected=(), min_recall=1.0, fix_policy="silent", repo="O/R1", split="holdout", + ) # fmt: skip + guard = Guard.build(gated_cases=[("https://github.com/o/r2", P["a"])], holdout_pairs=[holdout]) + kept, excluded = guard.apply(recs(false_alerts())) + assert kept == [] + assert excluded["holdout_pair"] == {f"github.com/o/r1@{P['a']}": 4} + assert excluded["gated_case"] == {f"github.com/o/r2@{P['a']}": 2} + [proposal] = propose(false_alerts(repos=("r1", "r2", "r3", "r3", "r4")), [_rule("noisy")], guard) + assert {e["repo"] for e in proposal.provenance["evidence"]} == {"github.com/o/r3", "github.com/o/r4"} + assert proposal.provenance["excluded"]["rows"] == 4 + + +# ---- provenance, the round, and outcomes --------------------------------------------------------------------------- + + +def _scaffold(tmp_path: Path) -> tuple[Path, Path, Any]: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("def f(x):\n return eval(x)\n") + rules_dir = tmp_path / "ruleset" + write_ruleset(rules_dir / "python" / "rules.toml", [_rule("good-eval", r"\beval\s*\("), _rule("noisy")]) + manifest = tmp_path / "bench.toml" + manifest.write_text( + 'name = "t"\nlanguage = "python_web"\n\n[source]\npath = "repo"\n\n[[expected]]\ncwe = "CWE-95"\n' + 'class = "code injection"\npath = "app.py"\nline = 2\nrule_id = "good-eval"\nsink = "eval"\nevidence = "eval"\n' + ) + return repo, rules_dir, load_benchmark_manifest(manifest) + + +def test_provenance_round_trips_journal_sidecar_store(tmp_path: Path) -> None: + store = FileStore(tmp_path / "memory") + store.ingest_rows("run-1", "t", false_alerts()) + records = store.rows() + [proposal] = propose_from_memory(records, {"noisy": _rule("noisy")}, {}, [], excluded=NO_GUARD, index_digest="idx") + repo, rules_dir, manifest = _scaffold(tmp_path) + journal, ledger = tmp_path / "cal" / "journal.json", tmp_path / "cal" / "rule_policy.json" + + outcome = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=rules_dir, proposals=[proposal]) + + assert outcome.accepted and read_rule_ledger(ledger)["noisy"]["status"] == "shadow" + [entry] = json.loads(journal.read_text())[0]["edits"] + assert entry["evidence"] == proposal.proposal_id and entry["rationale"] == f"memory:{proposal.proposal_id}" + side = read_sidecar(journal)[entry["evidence"]] + assert side["edit_key"] == entry["key"] and side["index_digest"] == "idx" and side["round"] == 1 + by_id = {rec.row["id"]: rec for rec in records} + for ev in side["evidence"]: # every cited row is in the store, at the object and version recorded + assert by_id[ev["id"]].key == ev["key"] and by_id[ev["id"]].version == ev["version"] + assert ev["version"].startswith("sha256:") + + rows = outcome_rows( + outcome.edits, outcome_of(outcome.reason, outcome.accepted) or "", 1, outcome.reason, {proposal.proposal_id: proposal}, "g" + ) + store.put_rows(rows) + back = store.rows(kind="proposal_outcome") + assert {r.row["outcome"] for r in back} == {"accepted"} and {r.row["proposal_id"] for r in back} == {proposal.proposal_id} + assert len(back) == 2 # one per (repository, pin) of the evidence + + again = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=rules_dir, proposals=[proposal]) + assert again.reason == "no_proposals" # the rule is shadow now: the proposal no longer applies + + +def test_a_reverted_memory_proposal_is_not_retried_on_the_same_evidence(tmp_path: Path) -> None: + repo, rules_dir, manifest = _scaffold(tmp_path) + rows = [verdict(r, P["a"], f"f{i}", "rejected") for i, r in enumerate(("r1", "r1", "r2"))] + rows += [alert("good-eval", r, P["a"], f"f{i}") for i, r in enumerate(("r1", "r1", "r2"))] + [proposal] = propose(rows, [_rule("good-eval", r"\beval\s*\(")]) + journal = tmp_path / "j.json" + first = run_round(repo, manifest, ledger_path=tmp_path / "l.json", journal_path=journal, ruleset_dir=rules_dir, proposals=[proposal]) + assert first.reason == "reverted" and outcome_of(first.reason, first.accepted) == "reverted" # shadowing drops recall + second = run_round(repo, manifest, ledger_path=tmp_path / "l.json", journal_path=journal, ruleset_dir=rules_dir, proposals=[proposal]) + assert second.reason == "no_proposals" + assert outcome_of("validation_failed: x", False) == "rejected" and outcome_of("no_proposals", False) is None + + +def test_default_round_journal_has_no_evidence_and_no_sidecar(tmp_path: Path) -> None: + repo, rules_dir, manifest = _scaffold(tmp_path) + (repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g(y):\n exec(y)\n") + journal = tmp_path / "j.json" + outcome = run_round(repo, manifest, ledger_path=tmp_path / "l.json", journal_path=journal, ruleset_dir=rules_dir) + assert outcome.edits # the benchmark's own edit + [entry] = json.loads(journal.read_text())[0]["edits"] + assert set(entry) == {"key", "lever", "rule_id", "from", "to", "rationale"} + assert not (tmp_path / SIDECAR).exists() + + +# ---- the CLI ------------------------------------------------------------------------------------------------------- + + +def test_default_improve_output_matches_the_recorded_baseline(capsys: pytest.CaptureFixture[str]) -> None: + root = Path(__file__).resolve().parents[1] + baseline = root / "benchmarks/measurements/2026-09-30-harnessx-removal-baseline/improve/stdout.txt" + manifest = "benchmarks/manifests/java-spring-boot-vulnerable.toml" + import os + + cwd = Path.cwd() + os.chdir(root) + try: + assert main(["improve", manifest, "--dry-run", "--no-pair-gate"]) == 0 + finally: + os.chdir(cwd) + assert capsys.readouterr().out.replace(str(root), "") == baseline.read_text() + + +def test_improve_memory_with_an_empty_store_has_no_proposals(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + repo, rules_dir, _ = _scaffold(tmp_path) + store = tmp_path / "memory" + store.mkdir() + (store / "index.jsonl").write_text("") + catalog = Path(__file__).resolve().parents[1] / "benchmarks/pairs/catalog.toml" + args = ["improve", str(tmp_path / "bench.toml"), "--ruleset-dir", str(rules_dir), "--dry-run", "--pair-catalog", str(catalog)] + assert main([*args, "--no-pair-gate", "--memory", str(store)]) == 0 + out = capsys.readouterr().out + assert "rows=0" in out and "memory_proposals=0" in out and "round 1: no_proposals" in out + with pytest.raises(SystemExit, match="no index.jsonl"): + main([*args, "--memory", str(tmp_path / "absent")])