#!/usr/bin/env python3 """Contract gates for co-writer suggestions. Reads a suggestions file — the JSON emitted by suggest.py, i.e. an object with a ``document`` string and a ``suggestions`` list — and enforces four blocking contracts. Each failure is named so a caller can act on it: span-minimality Every suggested_replacement edits only its own span: span.text matches document[start:end], the replacement differs from the span text, and it shares no leading or trailing whole word with the span (which would mean the span grabbed unchanged text and could be shrunk — the signature of a whole-sentence rewrite). replacement-scanner Every suggested_replacement passes both scanners in isolation AND introduces no new violation in context. accept-all Applying every replacement yields a document that passes both scanners with validate_preservation exit 0 vs the original. span-overlap Suggestion spans must not overlap. Exit 0 when all gates pass, 1 otherwise. Failures (with the offending gate) are reported as JSON on stdout. Usage: python3 scripts/check_suggestions.py suggestions.json python3 scripts/check_suggestions.py < suggestions.json """ from __future__ import annotations import argparse import json import sys from pathlib import Path from banned_phrase_scan import scan_for_violations from structure_scan import scan as structure_scan from validate_preservation import validate_preservation def _leading_shared_words(a: str, b: str) -> int: """Number of identical leading whole words shared by two strings.""" n = 0 for x, y in zip(a.split(), b.split()): if x == y: n += 1 else: break return n def _trailing_shared_words(a: str, b: str) -> int: ra = " ".join(reversed(a.split())) rb = " ".join(reversed(b.split())) return _leading_shared_words(ra, rb) def _line_starts(text: str) -> list[int]: starts = [0] for i, ch in enumerate(text): if ch == "\n": starts.append(i + 1) return starts def violation_spans(text: str) -> list[tuple[int, int]]: """Absolute (start, end) offsets of every banned-phrase/structural violation.""" starts = _line_starts(text) out = [] for v in scan_for_violations(text): s = starts[v["line_number"] - 1] + v["column"] - 1 out.append((s, s + len(v["phrase"]))) return out def _scanners_clean(text: str) -> bool: return not scan_for_violations(text) and not structure_scan(text).get("flags") def apply_all(document: str, suggestions: list[dict]) -> str: """Apply every non-null replacement, right-to-left so offsets stay valid.""" out = document for s in sorted(suggestions, key=lambda s: s["span"]["start"], reverse=True): rep = s.get("suggested_replacement") if rep is None: continue st, en = s["span"]["start"], s["span"]["end"] out = out[:st] + rep + out[en:] return out def check(document: str, suggestions: list[dict]) -> list[dict]: failures: list[dict] = [] # span-overlap: spans must be disjoint. order = sorted(range(len(suggestions)), key=lambda i: (suggestions[i]["span"]["start"], suggestions[i]["span"]["end"])) last_end = None last_i = None for i in order: st, en = suggestions[i]["span"]["start"], suggestions[i]["span"]["end"] if last_end is not None and st < last_end: failures.append({ "gate": "span-overlap", "suggestions": [last_i, i], "detail": f"span {st}-{en} overlaps the previous span ending at {last_end}", }) last_end, last_i = en, i # span-minimality: span accuracy + tight, local replacements. for i, s in enumerate(suggestions): st, en = s["span"]["start"], s["span"]["end"] span_text = document[st:en] if span_text != s["span"]["text"]: failures.append({ "gate": "span-minimality", "suggestion": i, "detail": "span.text does not match document[start:end]", }) rep = s.get("suggested_replacement") if rep is None: continue if rep == span_text: failures.append({ "gate": "span-minimality", "suggestion": i, "detail": "replacement is identical to the span text (no change)", }) continue if _leading_shared_words(span_text, rep) > 0 or _trailing_shared_words(span_text, rep) > 0: failures.append({ "gate": "span-minimality", "suggestion": i, "detail": "replacement shares leading/trailing whole words with the span; " "shrink the span so the edit is minimal", }) # replacement-scanner: each replacement is clean alone and in context. for i, s in enumerate(suggestions): rep = s.get("suggested_replacement") if rep is None: continue if not _scanners_clean(rep): failures.append({ "gate": "replacement-scanner", "suggestion": i, "detail": "replacement does not pass both scanners in isolation", }) st, en = s["span"]["start"], s["span"]["end"] ctx = document[:st] + rep + document[en:] new_start, new_end = st, st + len(rep) if any(a < new_end and new_start < b for (a, b) in violation_spans(ctx)): failures.append({ "gate": "replacement-scanner", "suggestion": i, "detail": "replacement introduces a violation in context", }) # accept-all: applying everything yields a clean, constraint-preserving doc. unresolved = [i for i, s in enumerate(suggestions) if s.get("suggested_replacement") is None] if unresolved: failures.append({ "gate": "accept-all", "detail": f"suggestions {unresolved} have no replacement; cannot accept-all", }) applied = apply_all(document, suggestions) if not _scanners_clean(applied): failures.append({ "gate": "accept-all", "detail": "the accept-all document still fails a scanner", }) preservation = validate_preservation(document, applied) if not preservation["passed"]: failures.append({ "gate": "accept-all", "detail": "validate_preservation failed against the original", "missing": preservation["missing"], }) return failures def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Contract gates for co-writer suggestions.") parser.add_argument( "path", nargs="?", help="Path to a suggestions JSON file (default: read stdin)" ) return parser.parse_args(argv) def main(argv: list[str]) -> int: args = parse_args(argv) if args.path: path = Path(args.path) if not path.exists(): print(f"Missing file: {path}", file=sys.stderr) return 2 raw = path.read_text(errors="replace") else: raw = sys.stdin.buffer.read().decode("utf-8", errors="replace") try: data = json.loads(raw) except json.JSONDecodeError as e: print(json.dumps({"passed": False, "error": f"invalid JSON: {e}"}, indent=2)) return 1 document = data.get("document") suggestions = data.get("suggestions", []) if not isinstance(document, str): print(json.dumps({"passed": False, "error": "missing 'document' string"}, indent=2)) return 1 failures = check(document, suggestions) result = { "passed": not failures, "failure_gates": sorted({f["gate"] for f in failures}), "failures": failures, } print(json.dumps(result, indent=2)) return 0 if not failures else 1 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))